From 3b3d796d28308ff1b34ed8928a5ecd5371cd06c3 Mon Sep 17 00:00:00 2001
From: Alan Liu <zaoxing@users.noreply.github.com>
Date: Fri, 11 Sep 2026 09:10:09 -0400
Subject: [PATCH 01/13] fix(catalog): decode \b and \f in the tuple parser,
 like the footer does

hydrate binds every catalog descriptor field against the pack footer, and
the two sides run different decoders: the catalog side parses ClickHouse's
rendered tuple, the footer side runs unquote_sql over sql_quote's map. The
tuple parser covered only \n \t \r \0 and let everything else through on
its deliberate `\'`/`\\` passthrough, so \b decoded to the letter "b" and
\f to "f" while the footer side decoded them to 0x08 and 0x0C. A hook_name
carrying either byte is legal -- _validate_text asks only for non-empty
UTF-8 under 512 bytes -- so it staged, uploaded and indexed, and then
hydrate refused it with "catalog descriptor does not match the pack
footer: field 12". Python reads over the native protocol with no TSV layer
and hydrates it fine.

The set is measured, not copied from the other switch: `SELECT
tuple(concat('block', char(N), 'resid')) FORMAT TSV` driven against a live
server over the whole control range shows ClickHouse escapes exactly
\0 \b \t \n \f \r \' \\ inside a tuple and leaves every other byte raw --
notably 0x07 and 0x0B, which sql_quote DOES rewrite (\a, \v). So only \b
and \f are added; the passthrough that keeps `\'` and `\\` from being
double-decoded stays as it is.

Pinned on the CPU gate by a new parametrized test that decodes each
measured rendering through `tuple_fields` and binds it against
`footer_row_fields`, and on the live gate by extending the hydrate
parametrization to the full measured set. Before the fix the CPU test
failed on bytes 8 and 12 and the live one on the same two with "field 12";
a full 1..127 sweep comparing both decoders now reports no disagreement.
---
 native/csrc/catalog/reader.cpp               | 18 ++++++++++
 tests/test_native_hydration_footer_fields.py | 36 ++++++++++++++++++++
 tests/test_native_reader_parity_live.py      | 12 +++++++
 3 files changed, 66 insertions(+)

diff --git a/native/csrc/catalog/reader.cpp b/native/csrc/catalog/reader.cpp
index e47a9c5b8..b7f93e41a 100644
--- a/native/csrc/catalog/reader.cpp
+++ b/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
         }
diff --git a/tests/test_native_hydration_footer_fields.py b/tests/test_native_hydration_footer_fields.py
index 83b76ebd1..b277458ab 100644
--- a/tests/test_native_hydration_footer_fields.py
+++ b/tests/test_native_hydration_footer_fields.py
@@ -203,3 +203,39 @@ def test_the_resolved_tuple_keeps_values_the_footer_binding_compares():
     assert _tuple_fields("('h',[2,8],0,'none')") == ["h", "[2,8]", "0", "none"]
     # A value that merely CONTAINS the token is untouched.
     assert _tuple_fields("('NULLABLE','a NULL b')") == ["NULLABLE", "a NULL b"]
+
+
+# --- the two decoders have to agree, byte for byte ---------------------------
+#
+# hydrate compares the tuple decoder's output (above) against the footer
+# decoder's (unquote_sql, sql_quote's map entry for entry). Where the two
+# disagree on ONE byte the binding refuses a legal capture with "catalog
+# descriptor does not match the pack footer: field N", and nothing but a
+# live hydrate of a capture carrying that byte could see it.
+#
+# The rendering below is MEASURED, not assumed: `SELECT tuple(concat('block',
+# char(N), 'resid')) FORMAT TSV` was driven against the live server for the
+# whole 1..127 range (plus char(0)). ClickHouse escapes exactly eight bytes
+# inside a tuple -- \0 \b \t \n \f \r \' \\ -- and leaves every other one
+# RAW, including 0x07 (BEL) and 0x0B (VT), which sql_quote does rewrite (as
+# \a and \v). So the two escape SETS are deliberately different, and it is
+# the decoded bytes, not the spellings, that have to match.
+TUPLE_RENDERING = {
+    0x00: "\\0", 0x07: "\x07", 0x08: "\\b", 0x09: "\\t", 0x0A: "\\n",
+    0x0B: "\x0b", 0x0C: "\\f", 0x0D: "\\r", 0x27: "\\'", 0x5C: "\\\\",
+}
+
+
+@pytest.mark.parametrize("byte", sorted(TUPLE_RENDERING))
+def test_the_tuple_and_footer_decoders_agree_on_every_byte_clickhouse_rewrites(byte):
+    """Both halves of the footer binding decode to the same bytes.
+
+    \\b and \\f decoded to the letters "b" and "f" on the catalog side while
+    the footer side decoded them to 0x08 and 0x0C, so a hook_name carrying
+    either (legal: _validate_text only asks for non-empty UTF-8 under 512
+    bytes) staged, uploaded and indexed, and then failed to hydrate.
+    """
+    value = "block" + chr(byte) + "resid"
+    (catalog,) = _tuple_fields(f"('block{TUPLE_RENDERING[byte]}resid')")
+    assert catalog == value
+    assert _fields(_quote(value), [catalog])[0]["matches"] is True
diff --git a/tests/test_native_reader_parity_live.py b/tests/test_native_reader_parity_live.py
index 8ad306dcd..83d3e2aab 100644
--- a/tests/test_native_reader_parity_live.py
+++ b/tests/test_native_reader_parity_live.py
@@ -2410,8 +2410,14 @@ def _native_read(driver, op, sel, fake_s3, **limits):
         insecure=True, **limits)
 
 
+# The list is the set of bytes ClickHouse actually escapes when it renders a
+# tuple, measured against this server with `SELECT tuple(concat('block',
+# char(N), 'resid')) FORMAT TSV` across the control range: \b \t \n \f \r \'
+# and \\ (plus \0, left out here because a NUL in an object key is a separate
+# question). 0x07 and 0x0B travel RAW inside a tuple and so need no entry.
 @pytest.mark.parametrize("hook_name", [
     "block\\resid", "block'quoted", "block\tresid",
+    "block\nresid", "block\rresid", "block\bresid", "block\fresid",
 ])
 def test_hydrate_accepts_metadata_the_sql_escaper_rewrites(fake_s3, hook_name):
     """The footer binding compares DECODED values, not escaped footer text.
@@ -2421,6 +2427,12 @@ def test_hydrate_accepts_metadata_the_sql_escaper_rewrites(fake_s3, hook_name):
     two representations refused these three valid captures with "catalog
     descriptor does not match the pack footer: field 12". Python hydrates
     them; native must too, and hand back the same 16 bytes.
+
+    Decoding is not enough on its own: the two decoders have to reach the
+    same BYTE. \\b and \\f decoded to the letters "b" and "f" on the catalog
+    side while the footer side decoded them to 0x08 and 0x0C, so a hook_name
+    carrying either staged, uploaded and indexed and then failed that same
+    field-12 binding.
     """
     import base64 as _base64
 

From b97daf7722f86a2337372ab3a7b3a41b04797593 Mon Sep 17 00:00:00 2001
From: Alan Liu <zaoxing@users.noreply.github.com>
Date: Fri, 11 Sep 2026 09:27:10 -0400
Subject: [PATCH 02/13] test(sink): pin the surrogate decoder on something the
 footer cannot undo
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

test_a_non_bmp_identifier_survives_the_metadata_decoder and
test_a_non_bmp_hook_name_round_trips were vacuous: both still passed with
the surrogate-pair combine branch compiled out (`if (false && ...)`,
rebuilt -- 2 passed and 26 passed). Both read hook_name back off the staged
pack, and the pack footer writer decodes every UTF-8 sequence and re-emits
\uXXXX, so the broken CESU-8 the decoder wrote (ED A0 BD ED B8 80) came
back out of the footer as exactly the 😀 a correct decoder
produces, and json.loads recombined it either way. The staged bytes were
traced: b':"block\\ud83d\\ude0', hook_name 'block😀', with the fix OFF. The
surrogate work was pinned only by a clickhouse/manual live test -- nothing
on the CPU gate.

So both now assert an observable the footer serializer cannot reach:

* the pack-sink test echoes the parsed RecordMetadata.hook_name BYTES back
  through a new session-less `parse_metadata` op on conformance_sink (the
  same shape conformance_catalog gained footer_row_fields/tuple_fields
  for), as hex, and compares against hook_name.encode("utf-8");
* the torch test gains a hook_name of exactly 512 UTF-8 bytes -- the limit
  _validate_text and ValidText both apply. CESU-8 is two bytes longer per
  non-BMP character, so at the limit the broken decoder is refused outright
  ("invalid capture metadata") where Python admits the record. The oracle
  assertion is now in the test, so the parity claim is explicit.

Measured with the combine branch disabled and rebuilt: the pack-sink test
fails with 626c6f636beda0bdedb880 != 626c6f636bf09f9880 and the torch test
with RuntimeError "invalid capture metadata"; both pass with it on.

Also drops tests/test_native_pack_sink.py:1137, which restated line 1136:
two equal `str`s necessarily have equal encodings, so
`staged.hook_name.encode("utf-8") == hook_name.encode("utf-8")` could not
fail. The real byte-level check is now the parse_metadata assertion above.
---
 native/csrc/sink/conformance_sink.cpp | 35 +++++++++++++++++++++++++++
 tests/test_native_adapter_torch.py    | 22 ++++++++++++++---
 tests/test_native_pack_sink.py        | 16 ++++++++++--
 3 files changed, 68 insertions(+), 5 deletions(-)

diff --git a/native/csrc/sink/conformance_sink.cpp b/native/csrc/sink/conformance_sink.cpp
index bb866ce82..cf0b2b793 100644
--- a/native/csrc/sink/conformance_sink.cpp
+++ b/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>
@@ -233,6 +236,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;
diff --git a/tests/test_native_adapter_torch.py b/tests/test_native_adapter_torch.py
index 2236fe0f8..2ee6a2daa 100644
--- a/tests/test_native_adapter_torch.py
+++ b/tests/test_native_adapter_torch.py
@@ -316,10 +316,26 @@ def test_a_fractional_shape_dimension_is_refused_not_truncated(tmp_path):
     assert sink.snapshot()["submitted_records"] == 0
 
 
-def test_a_non_bmp_hook_name_round_trips(tmp_path):
+@pytest.mark.parametrize("hook_name", [
+    "block\U0001F600",
+    # Exactly 512 UTF-8 bytes -- the limit _validate_text and ValidText both
+    # apply. This is the case the round-trip above cannot see: reading the
+    # name back off the pack proves nothing, because the footer writer decodes
+    # each UTF-8 sequence and re-emits \uXXXX, so a decoder that left the
+    # surrogate pair as two three-byte CESU-8 sequences produced a footer
+    # holding exactly the same 😀 a correct decoder does, and
+    # json.loads recombined it either way (measured: this test passed with the
+    # combine branch compiled out). The BYTE COUNT does differ -- 514 against
+    # 512 -- so at the limit the broken decoder is refused outright as
+    # "invalid capture metadata" where Python admits the record.
+    "b" * 508 + "\U0001F600",
+])
+def test_a_non_bmp_hook_name_round_trips(tmp_path, hook_name):
     """json.dumps writes U+1F600 as a surrogate pair; the sink must combine it."""
     meta = _meta(0)
-    meta["hook_name"] = "block\U0001F600"
+    meta["hook_name"] = hook_name
+    # The oracle admits both, so native must too.
+    assert CaptureMetadata.from_mapping(meta).hook_name == hook_name
     row = _row(meta, 0, 64, "float32")
     assert "\\ud83d\\ude00" in row["metadata_json"]
     sink, lease = _make_sink(tmp_path)
@@ -327,7 +343,7 @@ def test_a_non_bmp_hook_name_round_trips(tmp_path):
     assert sink.flush_and_wait(30.0)
     sink.rethrow_if_failed()
     ((staged, _),) = _read_staged(tmp_path)
-    assert staged.hook_name == "block\U0001F600"
+    assert staged.hook_name == hook_name
 
 
 def test_the_sink_derives_from_the_engines_record_sink(tmp_path):
diff --git a/tests/test_native_pack_sink.py b/tests/test_native_pack_sink.py
index 9bac1ad1b..b9c888ce7 100644
--- a/tests/test_native_pack_sink.py
+++ b/tests/test_native_pack_sink.py
@@ -1124,17 +1124,29 @@ def _read_staged_metadata(root):
 @pytest.mark.parametrize("hook_name", ["block\U0001F600", "block中文"])
 def test_a_non_bmp_identifier_survives_the_metadata_decoder(sink, tmp_path,
                                                              hook_name):
-    """A surrogate pair is ONE code point; the BMP name is the control."""
+    """A surrogate pair is ONE code point; the BMP name is the control.
+
+    The assertion that matters is on the DECODER's bytes, taken straight out
+    of `parse_metadata` before anything re-serializes them. Reading hook_name
+    back off the staged pack cannot see this bug at all: the footer writer
+    decodes each UTF-8 sequence and re-emits \\uXXXX, so the decoder's broken
+    CESU-8 (ED A0 BD ED B8 80) is written back out as a correct \\ud83d\\ude00
+    pair that json.loads recombines into 😀. Measured -- both this test and
+    test_native_adapter_torch's round-trip passed with the surrogate combine
+    branch compiled out.
+    """
     _open(sink, tmp_path / "spool")
     metadata = _row_meta(0, dtype="uint8", shape=[64], hook_name=hook_name)
     assert "\\ud83d\\ude00" in metadata or "\\u4e2d" in metadata, metadata
+    decoded = sink.call(op="parse_metadata", metadata_json=metadata)
+    assert decoded["ok"], decoded
+    assert decoded["hook_name_hex"] == hook_name.encode("utf-8").hex()
     response = _submit_row(sink, metadata, bytes(64), "uint8", [64])
     assert response["ok"], response
     assert sink.call(op="flush", timeout=30)["ok"]
     assert sink.call(op="close", timeout=30)["snapshot"]["persisted_records"] == 1
     (staged,) = _read_staged_metadata(tmp_path / "spool")
     assert staged.hook_name == hook_name
-    assert staged.hook_name.encode("utf-8") == hook_name.encode("utf-8")
 
 
 @pytest.mark.parametrize("field, value, reason", [

From cb1e49299d8972900ed4084e59f1a1ed7236eeca Mon Sep 17 00:00:00 2001
From: Alan Liu <zaoxing@users.noreply.github.com>
Date: Fri, 11 Sep 2026 09:29:41 -0400
Subject: [PATCH 03/13] fix(json): refuse a lone surrogate and a malformed
 \uXXXX, as the oracle does

The surrogate-pair combine handles a full pair, but a surrogate with NO
partner kept the three-byte form: \ud83d decoded to ED A0 BD and \udcff to
ED B3 BF. That is CESU-8, not UTF-8, and ValidText admits it -- it checks
lead/continuation byte SHAPE, not the surrogate range -- so the record was
packed and inserted. The oracle never gets that far: _validate_text calls
.encode("utf-8"), which raises UnicodeEncodeError ("surrogates not
allowed"), and no record lands. Measured through the driver, before:
hook_name_hex 626c6f636beda0bd / 626c6f636bedb3bf, ok:true.

The same lambda's hex4 did not validate its digits --
`h <= '9' ? h - '0' : (h | 0x20) - 'a' + 10` accepts any byte -- so \uZZZZ
decoded to U+25553 and was packed as F0 A5 95 93 (626c6f636bf0a59593,
ok:true), a name nothing in the request ever spelled. json.loads raises
"Invalid \uXXXX escape".

Where the refusal lives, and why: the decoder is the only place that can
SEE either fault (by the time the bytes exist the information is gone), but
it has no error channel and most of its callers read fields that nothing
validates, so it must stay total. It therefore does both: it emits U+FFFD,
so that whatever a caller does next no sequence that is not valid UTF-8
ever leaves it -- that alone ends the CESU-8-in-a-pack class -- and it
CLEARS an optional `bool* ok` so callers that do have an error channel can
refuse. The flag is only ever cleared, so one flag latches a whole object,
and an absent key leaves it alone. Both defaults are nullptr, so no
existing caller changes behaviour.

The two metadata parsers pass it and refuse: ParseMetadataJson
(record_row.cpp -- the production path, torch adapter through SubmitRow)
with "capture metadata text is not encodable UTF-8", and the driver's
ParseMetadata, which is the native side of CaptureMetadata.from_mapping,
on the same latch the out-of-range integer refusal already uses. dtype is
left on the plain decoder: a replacement character there simply is not one
of the fourteen dtype names and DtypeSupported already refuses it.

Red before green, against the unfixed binaries: all three new cases
answered ok:true from parse_metadata and were admitted by submit_row.
---
 native/csrc/common/json.cpp           | 77 +++++++++++++++++++++------
 native/csrc/common/json.h             | 15 +++++-
 native/csrc/sink/conformance_sink.cpp | 47 +++++++++++-----
 native/csrc/sink/record_row.cpp       | 40 +++++++++-----
 tests/test_native_pack_sink.py        | 50 +++++++++++++++++
 5 files changed, 188 insertions(+), 41 deletions(-)

diff --git a/native/csrc/common/json.cpp b/native/csrc/common/json.cpp
index 973fd4f2b..5e16fb79c 100644
--- a/native/csrc/common/json.cpp
+++ b/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,19 +26,35 @@ 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);
           q += 5;
+          if (first < 0) {
+            reject();
+            break;
+          }
+          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
@@ -35,16 +62,33 @@ std::string Unescape(const std::string& text, size_t& q) {
           // 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,7 +117,7 @@ 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;
@@ -81,8 +125,11 @@ std::string FindString(const std::string& text, const std::string& key,
     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 "";
 }
 
diff --git a/native/csrc/common/json.h b/native/csrc/common/json.h
index 1717b53be..daca7c21a 100644
--- a/native/csrc/common/json.h
+++ b/native/csrc/common/json.h
@@ -17,11 +17,22 @@ namespace dmi_common {
 // 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);
+//
+// `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 {
diff --git a/native/csrc/sink/conformance_sink.cpp b/native/csrc/sink/conformance_sink.cpp
index cf0b2b793..ada65833b 100644
--- a/native/csrc/sink/conformance_sink.cpp
+++ b/native/csrc/sink/conformance_sink.cpp
@@ -47,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
@@ -65,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"));
@@ -164,8 +177,14 @@ int main() {
                    &out);
     std::cout << out << "}\n";
   };
+  const auto refuse_bad_text = [] {
+    std::string out = "{\"ok\":false,\"what\":";
+    jc::EscapeJson("capture metadata text is not encodable UTF-8", &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;
@@ -282,6 +301,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 =
diff --git a/native/csrc/sink/record_row.cpp b/native/csrc/sink/record_row.cpp
index 7c8f788fe..b811d5fc5 100644
--- a/native/csrc/sink/record_row.cpp
+++ b/native/csrc/sink/record_row.cpp
@@ -37,22 +37,38 @@ 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) {
+    return fail("capture metadata text is not encodable UTF-8");
+  }
   // Presence first: layer_number == -1 is legal (logits-style captures),
   // so FindInt's missing sentinel cannot stand in for absence.
   for (const char* name :
diff --git a/tests/test_native_pack_sink.py b/tests/test_native_pack_sink.py
index b9c888ce7..84a79ca2f 100644
--- a/tests/test_native_pack_sink.py
+++ b/tests/test_native_pack_sink.py
@@ -1149,6 +1149,56 @@ def test_a_non_bmp_identifier_survives_the_metadata_decoder(sink, tmp_path,
     assert staged.hook_name == hook_name
 
 
+@pytest.mark.parametrize("hook_name", ["block\ud83d", "block\udcff"])
+def test_a_lone_surrogate_is_refused_not_persisted_as_cesu8(sink, tmp_path,
+                                                            hook_name):
+    """A surrogate with no partner has no UTF-8 encoding at all.
+
+    The oracle cannot even build the record: _validate_text calls
+    .encode("utf-8"), which raises UnicodeEncodeError. Native kept the
+    three-byte form -- \\ud83d became ED A0 BD -- which ValidText admits,
+    because it checks lead/continuation byte SHAPE and not the surrogate
+    range, so the record was packed and inserted as CESU-8.
+    """
+    mapping = _meta(0, dtype="uint8", shape=(64,)).to_mapping()
+    mapping["hook_name"] = hook_name
+    with pytest.raises(PackFormatError, match="surrogates not allowed"):
+        CaptureMetadata.from_mapping(mapping)
+
+    _open(sink, tmp_path / "spool")
+    metadata = _row_meta(0, dtype="uint8", shape=[64], hook_name=hook_name)
+    assert "\\ud83d" in metadata or "\\udcff" in metadata, metadata
+    assert sink.call(op="parse_metadata", metadata_json=metadata)["ok"] is False
+    assert _submit_row(sink, metadata, bytes(64), "uint8", [64])["ok"] is False
+    # The mapping path is the native side of CaptureMetadata.from_mapping and
+    # refuses it too, rather than admitting a name the oracle cannot hold.
+    refused = sink.call(op="submit", metadata=json.loads(metadata),
+                        payload_b64=base64.b64encode(bytes(64)).decode())
+    assert refused["ok"] is False, refused
+    assert sink.call(op="snapshot")["snapshot"]["submitted_records"] == 0
+
+
+def test_a_malformed_unicode_escape_is_refused_not_decoded(sink, tmp_path):
+    """\\uZZZZ has no code point; json.loads raises "Invalid \\uXXXX escape".
+
+    The four digits were never checked -- `h <= '9' ? h - '0' : (h | 0x20) -
+    'a' + 10` accepts anything -- so \\uZZZZ decoded to U+25553 and the field
+    was packed as F0 A5 95 93, a name nothing in the request ever spelled.
+    """
+    _open(sink, tmp_path / "spool")
+    valid = _row_meta(0, dtype="uint8", shape=[64])
+    assert '"hook_name": "h"' in valid, valid
+    # A single backslash in the metadata TEXT: json.dumps doubles it on the
+    # wire, and the driver's outer FindString undoes exactly that doubling.
+    metadata = valid.replace('"hook_name": "h"', '"hook_name": "block\\uZZZZ"')
+    with pytest.raises(ValueError):
+        json.loads(metadata)
+
+    assert sink.call(op="parse_metadata", metadata_json=metadata)["ok"] is False
+    assert _submit_row(sink, metadata, bytes(64), "uint8", [64])["ok"] is False
+    assert sink.call(op="snapshot")["snapshot"]["submitted_records"] == 0
+
+
 @pytest.mark.parametrize("field, value, reason", [
     ("layer_number", 0.5, "not an integer"),
     ("captured_at_ns", 123.5, "not an integer"),

From 93751642812feb6267df4fd9bf5a76cf362b061c Mon Sep 17 00:00:00 2001
From: Alan Liu <zaoxing@users.noreply.github.com>
Date: Fri, 11 Sep 2026 09:55:33 -0400
Subject: [PATCH 04/13] test(ring-e2e): assert WHICH transport carried each
 record

The docstring claimed records flowing through the Ring and a comment
labelled the second record CPU-direct, but persisted_records == 3,
failures == 0 and payload equality are identical whether one record took
the Ring or all three went CPU-direct. RecordRuntime.emit_output already
returns the answer -- StepReservation.OVERSIZED exactly on the CPU-direct
branch -- and the hook was discarding it. Keep it and pin the sequence
[RESERVED, OVERSIZED, RESERVED]: measured, and a records.py forced down
the CPU-direct branch for every record now fails the test where it used
to pass.
---
 tests/test_native_sink_ring_e2e.py | 32 +++++++++++++++++++++++++++---
 1 file changed, 29 insertions(+), 3 deletions(-)

diff --git a/tests/test_native_sink_ring_e2e.py b/tests/test_native_sink_ring_e2e.py
index d5c866712..a44f14f46 100644
--- a/tests/test_native_sink_ring_e2e.py
+++ b/tests/test_native_sink_ring_e2e.py
@@ -50,6 +50,13 @@ class _CaptureHookRuntime:
     def __init__(self, runtime) -> None:
         self._runtime = runtime
         self.metadata = None
+        # RecordRuntime.emit_output's return value is the ONLY observable
+        # that says which transport carried each record: it is OVERSIZED
+        # exactly on the CPU-direct branch (records.py:333) and a ring
+        # reservation otherwise. Persisted counts and payload bytes are
+        # identical either way, so without this the test cannot tell a
+        # record that went through the Ring from one that bypassed it.
+        self.reservations = []
 
     def should_emit(self, hook):
         return True
@@ -62,7 +69,9 @@ def prepare_output(self, *, hook, output_index, output_id, output_spec,
         entry = ProducerPlanBuilder().record_output(
             output_id=output_id, output_spec=output_spec, output=output,
         )
-        return self._runtime.emit_output(entry, self.metadata, output)
+        reservation = self._runtime.emit_output(entry, self.metadata, output)
+        self.reservations.append(reservation)
+        return reservation
 
 
 def _metadata(capture_id: str, tensor: torch.Tensor, *, step: int):
@@ -125,7 +134,14 @@ def test_native_pack_sink_is_the_engines_record_sink(tmp_path: Path):
 
 
 def test_records_flow_from_the_ring_into_a_native_pack(tmp_path: Path):
-    """The explicit entry point: record_sink=create_native_pack_sink(...)."""
+    """The explicit entry point: record_sink=create_native_pack_sink(...).
+
+    Three records, and the point is that they do NOT all take the same
+    route: two fit the 64 KiB payload ring and one does not. The persisted
+    count and the payload bytes are the same whichever transport ran, so
+    the per-record StepReservation is what pins the split.
+    """
+    from dmi.adapters.base import StepReservation
     from dmi.api.v1 import (
         HookPointV1, HookSpecV1, MonitoringEngine, TransportSpec,
     )
@@ -152,7 +168,8 @@ def test_records_flow_from_the_ring_into_a_native_pack(tmp_path: Path):
         hook_runtime.metadata = _metadata("capture-ring", first, step=0)
         hook(first.cuda())
 
-        # Above the 64 KiB Ring: the CPU-direct fallback, same sink.
+        # 128 KiB, above the 64 KiB Ring: the CPU-direct fallback, same
+        # sink. Asserted, not just asserted-in-a-comment, below.
         second = torch.arange(32 * 1024, dtype=torch.float32)
         hook_runtime.metadata = _metadata("capture-direct", second, step=1)
         hook(second.cuda())
@@ -167,6 +184,15 @@ def test_records_flow_from_the_ring_into_a_native_pack(tmp_path: Path):
         snapshot = handle.native_sink.snapshot()
         assert snapshot["persisted_records"] == 3, snapshot
         assert snapshot["failures"] == 0, snapshot
+        # 48 bytes, then 128 KiB, then 4 bytes against a 64 KiB payload
+        # ring: reserved in the ring, CPU-direct, reserved in the ring.
+        # RESERVED and not FLUSHED because neither ring record comes near
+        # filling the ring, so nothing forces a drain.
+        assert hook_runtime.reservations == [
+            StepReservation.RESERVED,
+            StepReservation.OVERSIZED,
+            StepReservation.RESERVED,
+        ], hook_runtime.reservations
     finally:
         engine.close()
 

From b1ba92a139ac673b18f9ee3c44d68e8914fbe895 Mon Sep 17 00:00:00 2001
From: Alan Liu <zaoxing@users.noreply.github.com>
Date: Fri, 11 Sep 2026 09:55:41 -0400
Subject: [PATCH 05/13] test(adapter): pin the exact metadata refusal, not the
 word "metadata"

match="metadata" is satisfied by every RowStatusName(kBadMetadata)
message -- "missing an integer field", "not an integer", "out of
range", "failed validation" -- and by the unrelated "descriptor
requires metadata JSON followed by one payload slice". Parametrise the
reason and compare the whole message, the way
tests/test_native_pack_sink.py already does. step_number=-1 must land on
OUT OF RANGE specifically: with the kOutOfRange answer folded into
not-an-integer the old test still passed and the new one does not.
---
 tests/test_native_adapter_torch.py | 33 +++++++++++++++++++++++-------
 1 file changed, 26 insertions(+), 7 deletions(-)

diff --git a/tests/test_native_adapter_torch.py b/tests/test_native_adapter_torch.py
index 2ee6a2daa..9208bc2a9 100644
--- a/tests/test_native_adapter_torch.py
+++ b/tests/test_native_adapter_torch.py
@@ -274,16 +274,25 @@ def test_a_scalar_envelope_is_admitted(tmp_path):
     assert payload == torch.tensor(3.5, dtype=torch.float32).numpy().tobytes()
 
 
-@pytest.mark.parametrize("field, value", [
-    ("layer_number", 0.5),
-    ("captured_at_ns", 123.5),
-    ("step_number", -1),
+@pytest.mark.parametrize("field, value, reason", [
+    ("layer_number", 0.5, "capture metadata field is not an integer"),
+    ("captured_at_ns", 123.5, "capture metadata field is not an integer"),
+    ("step_number", -1, "capture metadata integer is out of range"),
 ])
 def test_invalid_numeric_metadata_is_refused_not_converted(tmp_path, field,
-                                                            value):
+                                                            value, reason):
     """What CaptureMetadata refuses, the sink refuses -- nothing is persisted.
 
     These used to be persisted as 0, 123 and 18446744073709551615.
+
+    The exact `reason` is pinned, not just the word "metadata": every
+    metadata refusal carries RowStatusName(kBadMetadata) == "invalid capture
+    metadata", so matching on "metadata" alone would be satisfied by any
+    other refusal -- including the unrelated "descriptor requires metadata
+    JSON followed by one payload slice". The -1 case in particular has to
+    land on OUT OF RANGE, which is what the kUnsigned domain buys: without
+    it step_number=-1 parses fine and is packed as 18446744073709551615.
+    (Same style as tests/test_native_pack_sink.py's parametrised `reason`.)
     """
     from dmi.storage.capture.model import CaptureStorageError
 
@@ -292,11 +301,14 @@ def test_invalid_numeric_metadata_is_refused_not_converted(tmp_path, field,
     with pytest.raises((CaptureStorageError, ValueError, TypeError)):
         CaptureMetadata.from_mapping(meta)
     sink, lease = _make_sink(tmp_path)
-    with pytest.raises(RuntimeError, match="metadata"):
+    with pytest.raises(RuntimeError) as refusal:
         sink.submit_envelope(
             LAYOUT, [_row(meta, 0, 64, "float32")],
             torch.zeros(16, dtype=torch.float32),
         )
+    assert str(refusal.value) == (
+        f"NativePackSink: invalid capture metadata: {reason}"
+    ), (field, value, str(refusal.value))
     assert sink.snapshot()["submitted_records"] == 0
 
 
@@ -308,11 +320,18 @@ def test_a_fractional_shape_dimension_is_refused_not_truncated(tmp_path):
     with pytest.raises((CaptureStorageError, ValueError, TypeError)):
         CaptureMetadata.from_mapping(meta)
     sink, lease = _make_sink(tmp_path)
-    with pytest.raises(RuntimeError, match="metadata"):
+    # The specific refusal, not merely "metadata": a truncated 16.5 would
+    # have been admitted as (16,), matching the envelope shape, so the
+    # message that has to come back is the shape parser's own.
+    with pytest.raises(RuntimeError) as refusal:
         sink.submit_envelope(
             LAYOUT, [_row(meta, 0, 64, "float32", shape=(16,))],
             torch.zeros(16, dtype=torch.float32),
         )
+    assert str(refusal.value) == (
+        "NativePackSink: invalid capture metadata: "
+        "capture shape must be an integer list"
+    ), str(refusal.value)
     assert sink.snapshot()["submitted_records"] == 0
 
 

From de3671782216d95a51c8dca15cc51bba9d49e618 Mon Sep 17 00:00:00 2001
From: Alan Liu <zaoxing@users.noreply.github.com>
Date: Fri, 11 Sep 2026 09:56:36 -0400
Subject: [PATCH 06/13] test(sink-loader): assert the module the loader
 returns, not just non-None

_load_native_sink_extension either raises or returns
_load_named_extension's result, so `is not None` held by construction
and would have held for any other object. Check
RING_TYPES_ARE_STANDINS the way the two tests above it do: a loader
patched to hand back a different object passes the old assertion and
fails the new one.
---
 tests/test_engine_runtime_api.py | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/tests/test_engine_runtime_api.py b/tests/test_engine_runtime_api.py
index e42069d1b..eaf9e7862 100644
--- a/tests/test_engine_runtime_api.py
+++ b/tests/test_engine_runtime_api.py
@@ -780,4 +780,9 @@ def test_the_sink_loader_tolerates_a_transport_module_without_the_loader(
     )
     monkeypatch.setitem(sys.modules, "dmi.transport.native", fake)
     monkeypatch.setattr(dmi.transport, "native", fake, raising=False)
-    assert native_sink._load_native_sink_extension() is not None
+    # The module the loader returns, not merely "not None": the loader can
+    # only raise or hand back _load_named_extension's result, so `is not
+    # None` was satisfied by construction and would have held for any other
+    # object too. Same assertion style as the two tests above.
+    assert native_sink._load_native_sink_extension().RING_TYPES_ARE_STANDINS \
+        is True

From b99cd40e13c4746da6e699ca362e4a1289586efd Mon Sep 17 00:00:00 2001
From: Alan Liu <zaoxing@users.noreply.github.com>
Date: Fri, 11 Sep 2026 09:58:21 -0400
Subject: [PATCH 07/13] test(hydration): derive the footer width from
 CAPTURE_COLUMNS

32 is len(CAPTURE_COLUMNS) - 1, right today and silent tomorrow: adding a
capture column left the hand-built row at 32 tokens, the assertion at 32,
and hydration.cpp's matching `fields.size() != 32` guard drifting with
nothing to catch it. Assemble the row per column name and take the width
and the five spot-checked positions from the oracle. Proved on a copy
with a 33rd column: the derived test names the unrendered column, the
hard-coded one still passed.
---
 tests/test_native_hydration_footer_fields.py | 86 ++++++++++++++++----
 1 file changed, 68 insertions(+), 18 deletions(-)

diff --git a/tests/test_native_hydration_footer_fields.py b/tests/test_native_hydration_footer_fields.py
index b277458ab..cea463856 100644
--- a/tests/test_native_hydration_footer_fields.py
+++ b/tests/test_native_hydration_footer_fields.py
@@ -1,8 +1,9 @@
 """The footer-binding decoder: rendered footer rows against decoded catalog text.
 
 hydrate binds every catalog descriptor to the pack footer by comparing the
-32 fields of the footer's rendered VALUES row (pack_index's renderer, SQL
-escaping and all) against the catalog row the reader returns (TSV escapes
+fields of the footer's rendered VALUES row -- one per CAPTURE_COLUMNS entry
+bar index_version (pack_index's renderer, SQL
+escaping and all) -- against the catalog row the reader returns (TSV escapes
 already undone, a NULL arriving as the empty string). The two sides only
 compare correctly in ONE representation, so the decoder is pinned here on
 the CPU gate through the driver's session-less `footer_row_fields` op:
@@ -26,9 +27,15 @@
 
 import pytest
 
+from dmi.storage.capture.clickhouse_schema import CAPTURE_COLUMNS
+
 REPO_ROOT = Path(__file__).resolve().parents[1]
 DRIVER = REPO_ROOT / "native" / "build" / "conformance_catalog"
 
+# The footer's VALUES row is every capture column but the trailing
+# index_version, which the indexer supplies rather than the pack.
+FOOTER_COLUMNS = CAPTURE_COLUMNS[:-1]
+
 pytestmark = [
     pytest.mark.cpu,
     pytest.mark.skipif(
@@ -122,24 +129,67 @@ def test_uuid_literals_arrays_and_numbers_split_at_top_level():
     assert all(f["matches"] for f in fields), fields
 
 
-def test_a_full_32_field_row_decodes_to_32_fields():
-    """A rank-2 shape's inner comma must not split the row."""
-    strings = ["capture-0", "t", "e", "r", "s", "q", "n", "m", "mr"]
-    row = ",".join(
-        [_quote(s) for s in strings]
-        + ["NULL", _quote("v"), _quote("block\\resid"), "3", "0", "0", "0",
-           "1", "0", _quote("float32"), "[2,8]", "1700000000000000000",
-           "toUUID('0190e8d0-4b2a-7c3e-9f00-0123456789ab')", _quote("local"),
-           _quote("packs/a\\b.dmi-pack"), "4096", _quote("c" * 64), "1", "64",
-           "64", "64", _quote("none"), _quote("deadbeef")]
+def test_a_full_footer_row_decodes_to_one_field_per_footer_column():
+    """A rank-2 shape's inner comma must not split the row.
+
+    Width and positions come from CAPTURE_COLUMNS, the oracle this decoder
+    is a port of, not from a hand-counted 32. The footer carries every
+    capture column except the trailing index_version, and
+    hydration.cpp:708 guards on that same width -- so with the count
+    hard-coded, adding a column left this row at 32 tokens, the test
+    passing, and the native guard silently out of date. The row is
+    assembled per column name for the same reason: a new column has no
+    rendering here and says so by name.
+    """
+    pack_id = "0190e8d0-4b2a-7c3e-9f00-0123456789ab"
+    rendered = {
+        "capture_id": _quote("capture-0"),
+        "tenant_id": _quote("t"),
+        "experiment_id": _quote("e"),
+        "run_id": _quote("r"),
+        "session_id": _quote("s"),
+        "request_id": _quote("q"),
+        "sequence_id": _quote("n"),
+        "model_id": _quote("m"),
+        "model_revision": _quote("mr"),
+        "adapter_revision": "NULL",
+        "capture_policy_version": _quote("v"),
+        "hook_name": _quote("block\\resid"),
+        "layer_number": "3",
+        "producer_rank": "0",
+        "step_number": "0",
+        "token_start": "0",
+        "token_end": "1",
+        "batch_position": "0",
+        "dtype": _quote("float32"),
+        "shape": "[2,8]",
+        "captured_at_ns": "1700000000000000000",
+        "pack_id": f"toUUID('{pack_id}')",
+        "store_id": _quote("local"),
+        "object_key": _quote("packs/a\\b.dmi-pack"),
+        "object_bytes": "4096",
+        "pack_checksum": _quote("c" * 64),
+        "pack_record_count": "1",
+        "payload_offset": "64",
+        "stored_length": "64",
+        "decoded_length": "64",
+        "codec": _quote("none"),
+        "payload_checksum": _quote("deadbeef"),
+    }
+    assert sorted(rendered) == sorted(FOOTER_COLUMNS), (
+        sorted(set(FOOTER_COLUMNS) - set(rendered)),
+        sorted(set(rendered) - set(FOOTER_COLUMNS)),
     )
+
+    row = ",".join(rendered[column] for column in FOOTER_COLUMNS)
     fields = _fields(row)
-    assert len(fields) == 32, fields
-    assert fields[11]["text"] == "block\\resid"
-    assert fields[19]["text"] == "[2,8]"
-    assert fields[21]["text"] == "0190e8d0-4b2a-7c3e-9f00-0123456789ab"
-    assert fields[23]["text"] == "packs/a\\b.dmi-pack"
-    assert fields[9]["null"] is True
+    assert len(fields) == len(FOOTER_COLUMNS), fields
+    by_column = dict(zip(FOOTER_COLUMNS, fields))
+    assert by_column["hook_name"]["text"] == "block\\resid"
+    assert by_column["shape"]["text"] == "[2,8]"
+    assert by_column["pack_id"]["text"] == pack_id
+    assert by_column["object_key"]["text"] == "packs/a\\b.dmi-pack"
+    assert by_column["adapter_revision"]["null"] is True
 
 
 def test_a_mismatching_decoded_value_is_still_a_mismatch():

From 693a54b54b835f78d7647b11d9c110ebb95056e0 Mon Sep 17 00:00:00 2001
From: Alan Liu <zaoxing@users.noreply.github.com>
Date: Fri, 11 Sep 2026 10:46:21 -0400
Subject: [PATCH 08/13] fix(spool): account a ready file this object has not
 counted before

The retry and EEXIST-loser paths added nothing to the committed account, on
the reasoning that "the file was counted by the stage (or process start)
that created it". That holds only when the creator was THIS Spool object.
A ready file created by a SECOND Spool object on the same root after this
one's Open() has never been counted here, so adding nothing judged the cap
against 0. main's Recover/Scan split did not close this: the reconciliation
scan only runs when the capacity check has already failed, and the retry
path returns before any capacity check.

Executed, max_bytes=1500, two Spool objects on one root:
  before: other.Stage(pack1,1000)=ok; writer.Stage(pack1,1000) retry=ok,
          writer.Snapshot().bytes=0; writer.Stage(pack2,1000)=ok
          -> 2000 ready bytes on disk against a 1500 cap
  oracle: after the retry snapshot.bytes=1000, then SpoolFullError
          "2000 > 1500", 1000 bytes on disk
  after:  snapshot.bytes=1000, then kFull "2000 > 1500", 1000 on disk

Python makes the distinction with a SET: stage() runs both paths through
_account_ready_locked (spool.py:124 and :159), which is a no-op only when
the path is already in _accounted_ready. The aggregate alone cannot decide
it, so this ports that set -- accounted_ready_, path -> bytes -- and counts
each ready path at most once. It is rebuilt wholesale wherever the
committed aggregate it describes is: Open, the reconciliation scan in
Stage, and Scan (Recover/ListPending), so the two can never disagree.

Remove moves with it, and has to. Uncounting by staged.object_bytes while
the path stayed in the ledger would leave the ledger claiming a file that
is gone, and the next stage of that pack would then be treated as already
accounted and add nothing. Uncounting by PATH drops exactly the bytes this
object recorded, keeps main's property that a removal RETRY releases no
capacity (the path is already out of the ledger, so the retry is free), and
additionally matches the oracle, which unaccounts a path whose file has
vanished (spool.py:314) instead of leaving it charged.

What is deliberately unchanged: the reservation account, and main's
Scan/ListPending structure including its `inflight_temps_` skip. Only
committed_* and the new ledger move here. Re-verified by execution after
the change -- every Stage failure exit (kFull, kBadArgument, kIo, the
EEXIST loser's kConflict) still leaves reserved_* at zero and the cap's
room intact; a Recover across a held reservation rebuilds only the
committed half; Snapshot() is still committed + reserved; the
concurrent-admission case (A reserved and paused, B refused, 1000 on disk,
then 500 more admitted) still holds; 20 concurrent 2MiB stages against
1830 Recover() passes lost no in-flight temp (the same harness fails 20/20
with the inflight_temps_ skip removed). peak_bytes_ now also moves on the
retry path, which is what _account_ready_locked does.

Red first: the two new C++ cases -- a retry of another object's ready file,
and an EEXIST loser against another object's winner -- fail 8 checks against
main's spool.cpp, with "second pack was admitted: ok", "loser admitted a
second pack: ok" and 2000 bytes on disk. The existing case at
test_spool_reservations.cpp:211 could not see this: it retries a file the
same object staged, which is the half that was already right.
---
 native/csrc/store/spool.cpp              |  97 ++++++++++++++----
 native/csrc/store/spool.h                |  23 +++++
 tests/native/test_spool_reservations.cpp | 119 +++++++++++++++++++++--
 3 files changed, 215 insertions(+), 24 deletions(-)

diff --git a/native/csrc/store/spool.cpp b/native/csrc/store/spool.cpp
index 52a39c398..7027f84a0 100644
--- a/native/csrc/store/spool.cpp
+++ b/native/csrc/store/spool.cpp
@@ -262,11 +262,14 @@ SpoolStatus Spool::Open(SpoolConfig config, Spool* out, std::string* error) {
   out->committed_entries_ = 0;
   out->reserved_bytes_ = 0;
   out->reserved_entries_ = 0;
+  out->accounted_ready_.clear();
   out->inflight_temps_.clear();
   out->peak_bytes_ = 0;
   out->generation_ = 0;
   // Account pre-existing files exactly like the Python constructor: ready
-  // files plus stale .open files both count until Recover() runs.
+  // files plus stale .open files both count until Recover() runs, and only
+  // the ready PATHS are remembered (_accounted_ready = ready_bytes), so a
+  // later retry of one of them is recognised as already counted.
   for (const auto& entry :
        fs::recursive_directory_iterator(out->root_, ec)) {
     if (ec) break;
@@ -276,12 +279,36 @@ SpoolStatus Spool::Open(SpoolConfig config, Spool* out, std::string* error) {
     const bool is_open = HasSuffix(name, kOpenSuffix);
     if (!is_ready && !is_open) continue;
     out->committed_bytes_ += entry.file_size();
-    if (is_ready) ++out->committed_entries_;
+    if (is_ready) {
+      ++out->committed_entries_;
+      out->accounted_ready_.emplace(entry.path().string(), entry.file_size());
+    }
   }
   out->peak_bytes_ = out->committed_bytes_;
   return SpoolStatus::kOk;
 }
 
+bool Spool::AccountReadyLocked(const std::string& path,
+                               uint64_t object_bytes) {
+  if (accounted_ready_.count(path) != 0) return false;
+  accounted_ready_.emplace(path, object_bytes);
+  committed_bytes_ += object_bytes;
+  ++committed_entries_;
+  peak_bytes_ = std::max(peak_bytes_, committed_bytes_ + reserved_bytes_);
+  return true;
+}
+
+bool Spool::UnaccountReadyLocked(const std::string& path) {
+  const auto found = accounted_ready_.find(path);
+  if (found == accounted_ready_.end()) return false;
+  // Subtract what was RECORDED for this path, not what the caller believes it
+  // to be: the recorded value is the one that went into the aggregate.
+  if (committed_bytes_ >= found->second) committed_bytes_ -= found->second;
+  if (committed_entries_ > 0) --committed_entries_;
+  accounted_ready_.erase(found);
+  return true;
+}
+
 void Spool::SetStageHookForTesting(std::function<void()> hook) {
   std::lock_guard<std::mutex> lock(mutex_);
   stage_hook_for_testing_ = std::move(hook);
@@ -379,6 +406,7 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns,
         // reservation should have refused.
         uint64_t actual = 0;
         uint64_t ready_count = 0;
+        std::unordered_map<std::string, uint64_t> seen_ready;
         std::error_code walk_ec;
         for (const auto& entry :
              fs::recursive_directory_iterator(root_, walk_ec)) {
@@ -388,6 +416,7 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns,
           if (HasSuffix(name, kReadySuffix)) {
             actual += entry.file_size();
             ++ready_count;
+            seen_ready.emplace(entry.path().string(), entry.file_size());
           } else if (HasSuffix(name, kOpenSuffix) &&
                      inflight_temps_.count(entry.path().string()) == 0) {
             // Someone else's in-progress write (another process, or a
@@ -398,6 +427,9 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns,
         }
         committed_bytes_ = actual;
         committed_entries_ = ready_count;
+        // The path ledger is rebuilt with the aggregate it describes, so the
+        // two never disagree about which files the committed account holds.
+        accounted_ready_ = std::move(seen_ready);
         if (committed_bytes_ + reserved_bytes_ + n > max_bytes_) {
           if (error) {
             *error = "spool byte limit exceeded: " +
@@ -432,10 +464,16 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns,
     if (reserved_bytes_ >= n) reserved_bytes_ -= n;
     if (reserved_entries_ > 0) --reserved_entries_;
     inflight_temps_.erase(temp);
-    committed_bytes_ += n;
-    ++committed_entries_;
+    AccountReadyLocked(ready, n);
     ++generation_;
   };
+  // The retry and EEXIST-loser paths both end on a ready file that already
+  // exists. Whether it costs anything depends on whether THIS object has
+  // counted that path before, which only the path ledger can answer.
+  auto account_existing = [&] {
+    std::lock_guard<std::mutex> lock(mutex_);
+    if (AccountReadyLocked(ready, n)) ++generation_;
+  };
   auto fill_out = [&] {
     out->pack_id = pack_id;
     out->created_at_ns = created_at_ns;
@@ -468,9 +506,18 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns,
       if (error) *error = "spool contains different content: " + object_key;
       return SpoolStatus::kConflict;
     }
-    // A retry adds no accounting: the file was counted by the stage (or
-    // process start) that created it. Every successful retry still closes
-    // the durability window itself with a fresh fsync chain.
+    // A retry of a file THIS object created adds no accounting; a retry of
+    // one created by a second Spool object on the same root after this
+    // one's Open() adds its full size, because nothing here has counted it.
+    // Python makes exactly that distinction -- spool.py:124 runs the retry
+    // through _account_ready_locked, which is a no-op only for a path
+    // already in _accounted_ready. Adding nothing unconditionally judged the
+    // cap against 0: with max 1500, a 1000-byte file staged through another
+    // object and retried here still admitted another 1000.
+    //
+    // Every successful retry still closes the durability window itself with
+    // a fresh fsync chain.
+    account_existing();
     if (!FsyncChain(root_, dir, error)) return SpoolStatus::kIo;
     fill_out();
     return SpoolStatus::kOk;
@@ -493,15 +540,18 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns,
   }
   if (::link(temp.c_str(), ready.c_str()) != 0) {
     if (errno == EEXIST) {
-      // Lost the race: the winner's file is already counted (by its stage
-      // or by process start), so release this reservation, then validate
-      // the winner exactly like the retry path.
+      // Lost the race: release this reservation, then validate the winner
+      // exactly like the retry path. The winner's file is counted here only
+      // if this object has not counted that path already -- the winner may
+      // be another Spool object entirely, whose stage touched nothing in
+      // this account. Python does the same (spool.py:159).
       ::unlink(temp.c_str());
       unreserve();
       if (!validate_ready()) {
         if (error) *error = "spool contains different content: " + object_key;
         return SpoolStatus::kConflict;
       }
+      account_existing();
       // The winner may still be between link() and its own fsync: do not
       // acknowledge its dirent before independently making the chain
       // durable (mirrors the Python loser's fsync).
@@ -542,6 +592,7 @@ SpoolStatus Spool::Scan(std::vector<StagedPack>* out, bool discard_open_files,
   std::error_code ec;
   std::vector<std::string> readies;
   uint64_t bytes = 0;
+  std::unordered_map<std::string, uint64_t> seen_ready;
   for (const auto& entry :
        fs::recursive_directory_iterator(root_, ec)) {
     if (ec) break;
@@ -594,11 +645,16 @@ SpoolStatus Spool::Scan(std::vector<StagedPack>* out, bool discard_open_files,
     staged.object_bytes = size;
     out->push_back(std::move(staged));
     bytes += size;
+    seen_ready.emplace(path, size);
   }
   // Recovery rebuilds the committed account only; a stage in flight on
-  // another thread keeps its reservation.
+  // another thread keeps its reservation. The path ledger is rebuilt with
+  // it (_commit_recovery_locked does the same), so the surviving entries are
+  // exactly the ones a later retry will recognise as already counted, and
+  // the quarantined ones are simply absent.
   committed_bytes_ = bytes;
   committed_entries_ = out->size();
+  accounted_ready_ = std::move(seen_ready);
   peak_bytes_ = std::max(peak_bytes_, committed_bytes_ + reserved_bytes_);
   ++generation_;
   (void)error;
@@ -615,8 +671,12 @@ SpoolStatus Spool::Remove(const StagedPack& staged, std::string* error) {
         if (error) *error = "cannot inspect staged pack: " + ec.message();
         return SpoolStatus::kIo;
       }
-      // Removal retries must not release another pack's capacity. A stale
-      // count after an external removal is reconciled before refusing Stage.
+      // The file is already gone -- removed through another Spool object, or
+      // by an earlier call here. Drop it from this object's account only if
+      // this object was counting that PATH: a removal RETRY finds nothing
+      // recorded and costs nothing, so it cannot release another pack's
+      // capacity (which a blind subtraction did).
+      if (UnaccountReadyLocked(staged.path)) ++generation_;
       return SpoolStatus::kOk;
     }
     const std::string name = fs::path(staged.path).filename().string();
@@ -634,10 +694,13 @@ SpoolStatus Spool::Remove(const StagedPack& staged, std::string* error) {
       if (error) *error = "cannot remove staged pack: " + std::string(strerror(errno));
       return SpoolStatus::kIo;
     }
-    if (committed_bytes_ >= staged.object_bytes) {
-      committed_bytes_ -= staged.object_bytes;
-    }
-    if (committed_entries_ > 0) --committed_entries_;
+    // Uncount by PATH: the bytes that leave the aggregate are the ones this
+    // object recorded for it, and a path this object never counted (removed
+    // on behalf of another Spool object) costs nothing, exactly as
+    // _unaccount_ready_locked does. Uncounting by anything else would let a
+    // path stay in the ledger after its file is gone, and the next stage of
+    // the same pack would then be treated as already accounted.
+    UnaccountReadyLocked(staged.path);
     ++generation_;
     parent = fs::path(staged.path).parent_path().string();
   }
diff --git a/native/csrc/store/spool.h b/native/csrc/store/spool.h
index 051527b29..a56005c21 100644
--- a/native/csrc/store/spool.h
+++ b/native/csrc/store/spool.h
@@ -19,6 +19,7 @@
 #include <functional>
 #include <mutex>
 #include <string>
+#include <unordered_map>
 #include <unordered_set>
 #include <vector>
 
@@ -105,6 +106,19 @@ class Spool {
  private:
   SpoolStatus Scan(std::vector<StagedPack>* out, bool discard_open_files,
                   std::string* error);
+  // Count/uncount one ready path in the committed account, at most once each
+  // -- Python's _account_ready_locked / _unaccount_ready_locked. `mutex_`
+  // must be held. Both return whether they actually changed the account.
+  //
+  // The aggregate alone cannot decide this. A ready file reached by the retry
+  // or EEXIST-loser path may have been created by THIS object (already
+  // counted, so adding would double it) or by a second Spool object on the
+  // same root after this one's Open() (never counted, so adding nothing
+  // leaves the cap judged against 0 -- with max 1500, one 1000-byte file
+  // staged elsewhere then retried here left room for another 1000).
+  bool AccountReadyLocked(const std::string& path, uint64_t object_bytes);
+  bool UnaccountReadyLocked(const std::string& path);
+
   std::string root_;
   uint64_t max_bytes_ = 0;
   mutable std::mutex mutex_;
@@ -124,6 +138,15 @@ class Spool {
   uint64_t committed_entries_ = 0;
   uint64_t reserved_bytes_ = 0;
   uint64_t reserved_entries_ = 0;
+  // Which ready paths the committed account currently includes, and for how
+  // many bytes -- Python's _accounted_ready. It is what makes the accounting
+  // idempotent per PATH rather than per call, so the same ready file can be
+  // met more than once (retry, EEXIST loser, a scan) and be counted exactly
+  // once. Rebuilt wholesale wherever committed_* is. Stale .open bytes are
+  // in committed_bytes_ without being here, exactly as in Python's
+  // constructor, so this map is a ledger of ready paths, not a second copy
+  // of the aggregate.
+  std::unordered_map<std::string, uint64_t> accounted_ready_;
   std::unordered_set<std::string> inflight_temps_;
   uint64_t peak_bytes_ = 0;
   uint64_t generation_ = 0;
diff --git a/tests/native/test_spool_reservations.cpp b/tests/native/test_spool_reservations.cpp
index 5a399fa1a..e32b07d43 100644
--- a/tests/native/test_spool_reservations.cpp
+++ b/tests/native/test_spool_reservations.cpp
@@ -1,5 +1,5 @@
-// Spool capacity accounting under the two access patterns the counters have
-// to survive at once:
+// Spool capacity accounting under the access patterns the counters have to
+// survive at once:
 //
 //   1. A SECOND Spool object (an uploader in another thread or process) on
 //      the same root removes a ready file. The first object's committed
@@ -8,12 +8,18 @@
 //   2. While one stager holds a reservation but has written NOTHING yet,
 //      another stager trips that reconciliation. The scan sees an empty
 //      directory; it must not erase the first stager's reservation.
+//   3. A ready file created by a SECOND Spool object after this one's
+//      Open() is reached by this one's retry or EEXIST-loser path. This
+//      object's committed account has never counted it, so it has to count
+//      it now -- but exactly once, however many times that path is met.
 //
-// A single counter cannot satisfy both: replacing it with the scan fixes
-// (1) and breaks (2) -- with max 1500, A reserved 1000, B's 1000 scanned an
-// empty root, and both were admitted (2000 bytes on disk against a 1500
-// limit). Keeping committed bytes and in-flight reservations as separate
-// accounts, and letting the scan overwrite only the first, satisfies both.
+// A single counter cannot satisfy (1) and (2) at once: replacing it with
+// the scan fixes (1) and breaks (2) -- with max 1500, A reserved 1000, B's
+// 1000 scanned an empty root, and both were admitted (2000 bytes on disk
+// against a 1500 limit). Keeping committed bytes and in-flight reservations
+// as separate accounts, and letting the scan overwrite only the first,
+// satisfies both. Neither aggregate can decide (3) on its own, so the
+// committed account also carries the set of ready PATHS it includes.
 //
 // Built and run by tests/test_native_spool_reservations.py.
 
@@ -215,6 +221,103 @@ void TestARefusedStageLeavesNoReservationBehind() {
   CHECK(spool.Snapshot().entries == 1);
 }
 
+// (3) A ready file this object NEVER accounted for. The retry above is the
+// easy half: the object had already counted that file, so adding nothing is
+// right. When a SECOND Spool object on the same root creates the file after
+// this one's Open(), the retry path meets a file worth 1000 bytes that this
+// object's committed account has never seen -- and adding nothing there
+// leaves the cap judged against 0.
+//
+// Python takes the retry and the EEXIST-loser paths through
+// _account_ready_locked (spool.py:124 and :159), which is a no-op only when
+// the path is already in _accounted_ready. Its measured answer: the retry
+// snapshot reads 1000, and the next 1000-byte stage raises SpoolFullError
+// ("2000 > 1500") with 1000 bytes on disk.
+void TestARetryOfAnotherObjectsReadyFileIsAccounted() {
+  const std::string root = FreshRoot("foreign-retry");
+  dmi_store::SpoolConfig config{root, 1500};
+  dmi_store::Spool writer, other;
+  std::string error;
+  CHECK(dmi_store::Spool::Open(config, &writer, &error) ==
+        dmi_store::SpoolStatus::kOk);
+  CHECK(dmi_store::Spool::Open(config, &other, &error) ==
+        dmi_store::SpoolStatus::kOk);
+
+  // `other` writes the ready file; `writer` has never seen it.
+  dmi_store::StagedPack out;
+  CHECK(StageBytes(other, 1, 1000, &out, &error) ==
+        dmi_store::SpoolStatus::kOk);
+  CHECK(writer.Snapshot().bytes == 0);
+
+  // The same pack through `writer`: the ready file is already there, so this
+  // is the retry path. It succeeds -- the content matches -- and it has to
+  // leave the 1000 bytes IN the writer's account.
+  CHECK(StageBytes(writer, 1, 1000, &out, &error) ==
+        dmi_store::SpoolStatus::kOk);
+  CHECK(writer.Snapshot().bytes == 1000);
+  CHECK(writer.Snapshot().entries == 1);
+
+  // So a second, different pack of 1000 does not fit under the 1500 cap.
+  const dmi_store::SpoolStatus status =
+      StageBytes(writer, 2, 1000, &out, &error);
+  CHECK(status == dmi_store::SpoolStatus::kFull);
+  if (status != dmi_store::SpoolStatus::kFull) {
+    std::cerr << "second pack was admitted: "
+              << dmi_store::SpoolStatusName(status) << "\n";
+  }
+  CHECK(BytesOnDisk(root) == 1000);
+
+  // And a retry stays idempotent: running it twice more must not count the
+  // same path again.
+  CHECK(StageBytes(writer, 1, 1000, &out, &error) ==
+        dmi_store::SpoolStatus::kOk);
+  CHECK(StageBytes(writer, 1, 1000, &out, &error) ==
+        dmi_store::SpoolStatus::kOk);
+  CHECK(writer.Snapshot().bytes == 1000);
+  CHECK(writer.Snapshot().entries == 1);
+}
+
+// The EEXIST-loser path reaches the same file by a different route: both
+// objects write the same pack, and the one whose link() loses must still
+// end up with those bytes in its account.
+void TestAnEexistLoserAccountsForTheWinnersFile() {
+  const std::string root = FreshRoot("eexist-loser");
+  dmi_store::SpoolConfig config{root, 1500};
+  dmi_store::Spool loser, winner;
+  std::string error;
+  CHECK(dmi_store::Spool::Open(config, &loser, &error) ==
+        dmi_store::SpoolStatus::kOk);
+  CHECK(dmi_store::Spool::Open(config, &winner, &error) ==
+        dmi_store::SpoolStatus::kOk);
+
+  // The loser reserves, and while it is paused the winner links the ready
+  // file. The loser's own link() then fails with EEXIST -- the retry branch
+  // at the top of Stage ran before the file existed, so this is the only way
+  // to that code path.
+  dmi_store::StagedPack winner_out;
+  loser.SetStageHookForTesting([&] {
+    std::string inner;
+    CHECK(StageBytes(winner, 1, 1000, &winner_out, &inner) ==
+          dmi_store::SpoolStatus::kOk);
+  });
+  dmi_store::StagedPack out;
+  CHECK(StageBytes(loser, 1, 1000, &out, &error) ==
+        dmi_store::SpoolStatus::kOk);
+  loser.SetStageHookForTesting(nullptr);
+
+  CHECK(BytesOnDisk(root) == 1000);
+  CHECK(loser.Snapshot().bytes == 1000);
+  CHECK(loser.Snapshot().entries == 1);
+  const dmi_store::SpoolStatus status =
+      StageBytes(loser, 2, 1000, &out, &error);
+  CHECK(status == dmi_store::SpoolStatus::kFull);
+  if (status != dmi_store::SpoolStatus::kFull) {
+    std::cerr << "loser admitted a second pack: "
+              << dmi_store::SpoolStatusName(status) << "\n";
+  }
+  CHECK(BytesOnDisk(root) == 1000);
+}
+
 void TestRepeatedRemovalDoesNotReleaseAnotherPacksCapacity() {
   dmi_store::Spool spool;
   std::string error;
@@ -238,6 +341,8 @@ int main() {
   TestSerialRemoveThroughAnotherSpoolIsReconciled();
   TestReconciliationKeepsAnInflightReservation();
   TestARefusedStageLeavesNoReservationBehind();
+  TestARetryOfAnotherObjectsReadyFileIsAccounted();
+  TestAnEexistLoserAccountsForTheWinnersFile();
   TestRepeatedRemovalDoesNotReleaseAnotherPacksCapacity();
   if (g_failures != 0) {
     std::cerr << g_failures << " check(s) failed\n";

From 4540f78aa4690479ad6f42b80dd4855d79953c67 Mon Sep 17 00:00:00 2001
From: Alan Liu <zaoxing@users.noreply.github.com>
Date: Fri, 11 Sep 2026 10:48:23 -0400
Subject: [PATCH 09/13] test(spool): drive the release path the reservation
 test claimed to cover

TestARefusedStageLeavesNoReservationBehind named the release of a
reservation, but a refused stage returns kFull before reserved_bytes_
moves -- there is nothing to release, so replacing the whole unreserve
lambda with a no-op left that case printing ok. Split it: the refusal and
retry half keeps its (sound) idempotency claim under an honest name and
now also shows the refusal left room, and a new case drives the EEXIST
loser, the reachable path that really does reserve and then give it back.
Its observable is that a 500-byte stage under the 1500 cap is ADMITTED
after the loser released 1000, which the leak mutant refuses with
"2500 > 1500".
---
 tests/native/test_spool_reservations.cpp | 60 ++++++++++++++++++++++--
 1 file changed, 56 insertions(+), 4 deletions(-)

diff --git a/tests/native/test_spool_reservations.cpp b/tests/native/test_spool_reservations.cpp
index e32b07d43..4f9a6372e 100644
--- a/tests/native/test_spool_reservations.cpp
+++ b/tests/native/test_spool_reservations.cpp
@@ -199,9 +199,11 @@ void TestReconciliationKeepsAnInflightReservation() {
   CHECK(spool.Snapshot().bytes == 1500);
 }
 
-// A refused stage and a lost link() race both release their reservation
-// (the reserved account must return to zero, or the spool leaks capacity).
-void TestARefusedStageLeavesNoReservationBehind() {
+// A stage that the cap refuses, and a retry of a pack this object already
+// staged, both leave the accounts exactly where they were. The refusal is the
+// cheap half -- the capacity check returns before reserved_bytes_ moves, so
+// there is no reservation to give back.
+void TestARefusedStageAndARetryLeaveTheAccountsAlone() {
   const std::string root = FreshRoot("refused");
   dmi_store::SpoolConfig config{root, 1500};
   dmi_store::Spool spool;
@@ -219,6 +221,55 @@ void TestARefusedStageLeavesNoReservationBehind() {
         dmi_store::SpoolStatus::kOk);
   CHECK(spool.Snapshot().bytes == 1000);
   CHECK(spool.Snapshot().entries == 1);
+  // And the refusal really did leave room: the 500 bytes still under the cap
+  // are stageable.
+  CHECK(StageBytes(spool, 3, 500, &out, &error) ==
+        dmi_store::SpoolStatus::kOk);
+  CHECK(spool.Snapshot().bytes == 1500);
+}
+
+// The one reachable path that TAKES a reservation and then has to give it
+// back: the EEXIST loser. It reserves, loses link() to a second Spool object
+// on the same root, releases its reservation and accounts the winner's file
+// instead. A release that leaked leaves the bytes on disk unchanged -- the
+// file is there either way -- and shows up only where the reservation is
+// still charged: the snapshot, and the room left under the cap. So the
+// observable that matters is that a stage which FITS is still admitted.
+void TestALostLinkRaceReleasesItsReservation() {
+  const std::string root = FreshRoot("lost-link-release");
+  dmi_store::SpoolConfig config{root, 1500};
+  dmi_store::Spool loser, winner;
+  std::string error;
+  CHECK(dmi_store::Spool::Open(config, &loser, &error) ==
+        dmi_store::SpoolStatus::kOk);
+  CHECK(dmi_store::Spool::Open(config, &winner, &error) ==
+        dmi_store::SpoolStatus::kOk);
+
+  dmi_store::StagedPack winner_out;
+  loser.SetStageHookForTesting([&] {
+    std::string inner;
+    CHECK(StageBytes(winner, 1, 1000, &winner_out, &inner) ==
+          dmi_store::SpoolStatus::kOk);
+  });
+  dmi_store::StagedPack out;
+  CHECK(StageBytes(loser, 1, 1000, &out, &error) ==
+        dmi_store::SpoolStatus::kOk);
+  loser.SetStageHookForTesting(nullptr);
+  CHECK(loser.Snapshot().bytes == 1000);
+  CHECK(loser.Snapshot().entries == 1);
+
+  // 1000 committed under a 1500 cap: 500 fits, and only fits if the loser's
+  // reservation went back.
+  const dmi_store::SpoolStatus status =
+      StageBytes(loser, 2, 500, &out, &error);
+  CHECK(status == dmi_store::SpoolStatus::kOk);
+  if (status != dmi_store::SpoolStatus::kOk) {
+    std::cerr << "a stage that fits was refused: "
+              << dmi_store::SpoolStatusName(status) << " " << error << "\n";
+  }
+  CHECK(BytesOnDisk(root) == 1500);
+  CHECK(loser.Snapshot().bytes == 1500);
+  CHECK(loser.Snapshot().entries == 2);
 }
 
 // (3) A ready file this object NEVER accounted for. The retry above is the
@@ -340,7 +391,8 @@ void TestRepeatedRemovalDoesNotReleaseAnotherPacksCapacity() {
 int main() {
   TestSerialRemoveThroughAnotherSpoolIsReconciled();
   TestReconciliationKeepsAnInflightReservation();
-  TestARefusedStageLeavesNoReservationBehind();
+  TestARefusedStageAndARetryLeaveTheAccountsAlone();
+  TestALostLinkRaceReleasesItsReservation();
   TestARetryOfAnotherObjectsReadyFileIsAccounted();
   TestAnEexistLoserAccountsForTheWinnersFile();
   TestRepeatedRemovalDoesNotReleaseAnotherPacksCapacity();

From 0c5eaa9b845438673c9b5629a023077303fa51c3 Mon Sep 17 00:00:00 2001
From: Alan Liu <zaoxing@users.noreply.github.com>
Date: Fri, 18 Sep 2026 11:57:55 -0400
Subject: [PATCH 10/13] fix(json): validate a \uXXXX escape before advancing
 past its digits

Unescape stepped q over the four "digits" BEFORE hex4's verdict, so a
malformed short escape whose digits run into the field's real closing
quote -- "x\u12" -- stepped past that quote, and the while loop kept
copying the FOLLOWING JSON text (comma, separator, the next field's
key) into the field's value until the next quote. Every FindString
caller that passes no `ok` flag (pack_index.cpp's footer parsing,
reader.cpp's cursor "fh") silently received a value containing
structural JSON: measured, {"hook_name":"x\u1","model_id":"m"} decoded
hook_name to "x<U+FFFD>model_id". A \u12 at the very end of the buffer
left q strictly past text.size(), breaking json.h's contract that the
decoder returns with q on the closing quote (later accesses are
guarded, so no OOB read -- but the position was a lie).

On failure the decoder now consumes only the `u` and resumes at the
character after it: whatever failed hex4 is copied as literal text, the
real closing quote terminates the value, and the `ok` latch still
carries the refusal, so every already-pinned metadata refusal
(\uZZZZ, the lone surrogates) answers exactly as before. The
low-surrogate partner lookup already validated before its q += 6 and
needed no change; its break now lands on text the fixed path re-scans
safely, and the new alignment test covers that case too.

Red before green: tests/native/test_json_unescape.cpp (driven by
tests/test_native_json_unescape.py, the spool-reservations pattern)
failed 11 checks against the unfixed decoder -- the swallowed next-key
value, q past the closing quote in three shapes, and q == text.size()+2
on the truncated tail. The sink-boundary refusal test gains the short
\u12 variant alongside \uZZZZ.
---
 native/csrc/common/json.cpp         |  13 ++-
 native/csrc/common/json.h           |   3 +-
 tests/native/test_json_unescape.cpp | 172 ++++++++++++++++++++++++++++
 tests/test_native_json_unescape.py  |  61 ++++++++++
 tests/test_native_pack_sink.py      |  11 +-
 5 files changed, 256 insertions(+), 4 deletions(-)
 create mode 100644 tests/native/test_json_unescape.cpp
 create mode 100644 tests/test_native_json_unescape.py

diff --git a/native/csrc/common/json.cpp b/native/csrc/common/json.cpp
index 5e16fb79c..9b0229df9 100644
--- a/native/csrc/common/json.cpp
+++ b/native/csrc/common/json.cpp
@@ -49,11 +49,22 @@ std::string Unescape(const std::string& text, size_t& q, bool* ok) {
             return static_cast<long>(v);
           };
           const long first = hex4(q);
-          q += 5;
+          // 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
diff --git a/native/csrc/common/json.h b/native/csrc/common/json.h
index daca7c21a..742741b74 100644
--- a/native/csrc/common/json.h
+++ b/native/csrc/common/json.h
@@ -16,7 +16,8 @@ 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.
+// 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
diff --git a/tests/native/test_json_unescape.cpp b/tests/native/test_json_unescape.cpp
new file mode 100644
index 000000000..b1c196c51
--- /dev/null
+++ b/tests/native/test_json_unescape.cpp
@@ -0,0 +1,172 @@
+// Unescape's \uXXXX FAILURE path must leave the decoder on the string it is
+// decoding. hex4 validates the length and the four digits, but the advance
+// (`q += 5`) ran BEFORE the verdict was looked at, so a malformed escape
+// whose "digits" include the field's real closing quote stepped past that
+// quote, and the while loop kept copying the FOLLOWING JSON text -- comma,
+// separator, the next field's key -- into this field's value until the next
+// quote. Every FindString caller that passes no `ok` flag (pack_index.cpp's
+// footer parsing, reader.cpp's cursor "fh") silently received a value
+// containing structural JSON. A `\u12` at the very end of the buffer left
+// `q` strictly past text.size(), breaking json.h's contract that Unescape
+// "returns with `q` on the closing quote" (or at the end of an unterminated
+// text).
+//
+// On failure the decoder consumes only the `\u` and resumes at the
+// character after it: the characters that failed hex4 are copied as
+// literals, the real closing quote terminates the string, and the `ok`
+// latch still records the refusal.
+//
+// Built and run by tests/test_native_json_unescape.py.
+
+#include <iostream>
+#include <string>
+
+#include "common/json.h"
+
+namespace jc = dmi_common;
+
+namespace {
+
+int g_failures = 0;
+
+#define CHECK(cond)                                                        \
+  do {                                                                     \
+    if (!(cond)) {                                                         \
+      std::cerr << __FILE__ << ":" << __LINE__ << ": CHECK failed: " #cond \
+                << "\n";                                                   \
+      ++g_failures;                                                        \
+    }                                                                      \
+  } while (0)
+
+// U+FFFD, the replacement character the decoder emits for a refusal.
+const std::string kFFFD = "\xEF\xBF\xBD";
+
+// A short escape whose "digits" run into the closing quote, spaced
+// separator style (json.dumps's default). The refusal must latch AND the
+// decoder must stop on the field's real closing quote: the two characters
+// that failed hex4 come through as literals, and nothing past the quote --
+// no comma, no separator, no next key -- leaks into the value.
+void ShortEscapeStopsOnTheClosingQuote() {
+  const std::string text =
+      "{\"hook_name\": \"x\\u12\", \"model_id\": \"m\"}";
+  const size_t start = text.find("x\\u12");
+  size_t q = start;
+  bool ok = true;
+  const std::string value = jc::Unescape(text, q, &ok);
+  CHECK(!ok);
+  CHECK(value == "x" + kFFFD + "12");
+  CHECK(q == start + 5);  // the field's real closing quote
+  CHECK(q < text.size() && text[q] == '"');
+
+  // The no-ok FindString path, as pack_index.cpp and reader.cpp call it:
+  // the value must not contain structural JSON from beyond the quote.
+  const std::string hook = jc::FindString(text, "hook_name");
+  CHECK(hook.find(',') == std::string::npos);
+  CHECK(hook.find("model_id") == std::string::npos);
+
+  // A subsequent field in the same object still decodes correctly.
+  bool ok_model = true;
+  CHECK(jc::FindString(text, "model_id", 0, &ok_model) == "m");
+  CHECK(ok_model);
+}
+
+// Compact separators: the unconditional advance landed exactly on the
+// first character of the NEXT KEY, so the no-ok value came back as
+// "x�model_id" -- the neighbouring field's name, spelled by nothing
+// in the input's own value.
+void CompactShortEscapeDoesNotSwallowTheNextKey() {
+  const std::string text = "{\"hook_name\":\"x\\u1\",\"model_id\":\"m\"}";
+  const std::string hook = jc::FindString(text, "hook_name");
+  CHECK(hook == "x" + kFFFD + "1");
+  CHECK(hook.find("model_id") == std::string::npos);
+  CHECK(jc::FindString(text, "model_id") == "m");
+}
+
+// Sequential decoding: after a rejected escape in the FIRST field, `q` must
+// sit on that field's closing quote so later fields scanned from `q` stay
+// aligned with the object.
+void SequentialFieldsStayAlignedAfterARejectedEscape() {
+  const std::string text =
+      "{\"a\": \"p\\uZZ\", \"b\": \"second\", \"c\": \"third\"}";
+  const size_t start = text.find("p\\uZZ");
+  size_t q = start;
+  bool ok = true;
+  const std::string a = jc::Unescape(text, q, &ok);
+  CHECK(!ok);
+  CHECK(a == "p" + kFFFD + "ZZ");
+  CHECK(q == start + 5);  // on the closing quote of "a"'s value
+  CHECK(jc::FindString(text, "b", q) == "second");
+  CHECK(jc::FindString(text, "c", q) == "third");
+}
+
+// A truncated escape at the very end of the buffer: `q` must come to rest
+// AT text.size() (the unterminated-string exit), never past it.
+void TruncatedEscapeAtEndOfBufferKeepsQInBounds() {
+  {
+    std::string text = "\"x\\u12";  // opening quote, then x\u12, no close
+    size_t q = 1;
+    bool ok = true;
+    const std::string value = jc::Unescape(text, q, &ok);
+    CHECK(!ok);
+    CHECK(value == "x" + kFFFD + "12");
+    CHECK(q == text.size());
+  }
+  {
+    std::string text = "\"x\\u1";
+    size_t q = 1;
+    bool ok = true;
+    const std::string value = jc::Unescape(text, q, &ok);
+    CHECK(!ok);
+    CHECK(value == "x" + kFFFD + "1");
+    CHECK(q == text.size());
+  }
+}
+
+// A lead surrogate whose PARTNER escape is the malformed short one: the
+// lone-surrogate refusal fires first, then the second escape is decoded on
+// its own and refused too -- and the decoder still ends on the real
+// closing quote.
+void LeadSurrogateWithMalformedPartnerStaysAligned() {
+  const std::string text = "{\"k\": \"a\\ud83d\\u12\", \"m\": \"v\"}";
+  const size_t start = text.find("a\\ud83d");
+  size_t q = start;
+  bool ok = true;
+  jc::Unescape(text, q, &ok);
+  CHECK(!ok);
+  CHECK(q < text.size() && text[q] == '"');
+  CHECK(q == text.find("\", \"m\"", start));
+  CHECK(jc::FindString(text, "m", q) == "v");
+}
+
+// Valid escapes are untouched: BMP, a combined surrogate pair, and the
+// two-character escapes decode as before, `ok` stays latched true, and `q`
+// ends on the closing quote.
+void ValidEscapesAreUnchanged() {
+  const std::string text =
+      "{\"k\": \"a\\u2603\\ud83d\\ude00\\n\", \"m\": \"v\"}";
+  const size_t start = text.find("a\\u2603");
+  size_t q = start;
+  bool ok = true;
+  const std::string value = jc::Unescape(text, q, &ok);
+  CHECK(ok);
+  CHECK(value == "a\xE2\x98\x83\xF0\x9F\x98\x80\n");
+  CHECK(q < text.size() && text[q] == '"');
+  CHECK(jc::FindString(text, "m", q) == "v");
+}
+
+}  // namespace
+
+int main() {
+  ShortEscapeStopsOnTheClosingQuote();
+  CompactShortEscapeDoesNotSwallowTheNextKey();
+  SequentialFieldsStayAlignedAfterARejectedEscape();
+  TruncatedEscapeAtEndOfBufferKeepsQInBounds();
+  LeadSurrogateWithMalformedPartnerStaysAligned();
+  ValidEscapesAreUnchanged();
+  if (g_failures != 0) {
+    std::cerr << g_failures << " check(s) failed\n";
+    return 1;
+  }
+  std::cout << "ok\n";
+  return 0;
+}
diff --git a/tests/test_native_json_unescape.py b/tests/test_native_json_unescape.py
new file mode 100644
index 000000000..3109f2212
--- /dev/null
+++ b/tests/test_native_json_unescape.py
@@ -0,0 +1,61 @@
+"""Unescape's \\uXXXX failure path: the decoder must stay on the string.
+
+Compiles and runs tests/native/test_json_unescape.cpp against the real
+common/json.cpp. hex4 validates the four digits, but the advance (`q += 5`)
+ran BEFORE its verdict, so a malformed short escape whose "digits" include
+the field's real closing quote (`"x\\u12", ...`) stepped past that quote and
+the value silently swallowed following JSON text -- every FindString caller
+with no `ok` flag (pack footer parsing, the reader's cursor) received a
+value containing structural JSON -- and a `\\u12` at the end of the buffer
+left q strictly past text.size(), breaking json.h's "returns with `q` on
+the closing quote" contract.
+
+Needs a C++17 compiler (no libcrypto, unlike conformance_spool).
+"""
+
+from __future__ import annotations
+
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+
+@pytest.mark.cpu
+def test_unescape_failure_keeps_the_decoder_on_the_string(tmp_path):
+    compiler = shutil.which("g++") or shutil.which("c++")
+    if compiler is None:
+        pytest.skip("a C++17 compiler is required")
+
+    root = Path(__file__).resolve().parents[1]
+    csrc = root / "native" / "csrc"
+    source = root / "tests" / "native" / "test_json_unescape.cpp"
+    executable = tmp_path / "test_json_unescape"
+    compile_result = subprocess.run(
+        [
+            compiler,
+            "-std=c++17",
+            "-O0",
+            f"-I{csrc}",
+            str(source),
+            str(csrc / "common" / "json.cpp"),
+            "-o",
+            str(executable),
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert compile_result.returncode == 0, (
+        compile_result.stdout + compile_result.stderr
+    )
+
+    run_result = subprocess.run(
+        [str(executable)],
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=120,
+    )
+    assert run_result.returncode == 0, run_result.stdout + run_result.stderr
diff --git a/tests/test_native_pack_sink.py b/tests/test_native_pack_sink.py
index 84a79ca2f..4c7c122ab 100644
--- a/tests/test_native_pack_sink.py
+++ b/tests/test_native_pack_sink.py
@@ -1178,19 +1178,26 @@ def test_a_lone_surrogate_is_refused_not_persisted_as_cesu8(sink, tmp_path,
     assert sink.call(op="snapshot")["snapshot"]["submitted_records"] == 0
 
 
-def test_a_malformed_unicode_escape_is_refused_not_decoded(sink, tmp_path):
+@pytest.mark.parametrize("bad_name", ["block\\uZZZZ", "block\\u12"])
+def test_a_malformed_unicode_escape_is_refused_not_decoded(sink, tmp_path,
+                                                           bad_name):
     """\\uZZZZ has no code point; json.loads raises "Invalid \\uXXXX escape".
 
     The four digits were never checked -- `h <= '9' ? h - '0' : (h | 0x20) -
     'a' + 10` accepts anything -- so \\uZZZZ decoded to U+25553 and the field
     was packed as F0 A5 95 93, a name nothing in the request ever spelled.
+    \\u12 is the SHORT variant: its "digits" run into the field's closing
+    quote, which the decoder's unconditional `q += 5` used to step past, so
+    the value swallowed the following JSON text up to the next quote
+    (tests/native/test_json_unescape.cpp pins the alignment itself); the
+    refusal here must hold with the quote intact.
     """
     _open(sink, tmp_path / "spool")
     valid = _row_meta(0, dtype="uint8", shape=[64])
     assert '"hook_name": "h"' in valid, valid
     # A single backslash in the metadata TEXT: json.dumps doubles it on the
     # wire, and the driver's outer FindString undoes exactly that doubling.
-    metadata = valid.replace('"hook_name": "h"', '"hook_name": "block\\uZZZZ"')
+    metadata = valid.replace('"hook_name": "h"', f'"hook_name": "{bad_name}"')
     with pytest.raises(ValueError):
         json.loads(metadata)
 

From cd5d042c2fd5d2699b6dd8bc42579faa0eab956e Mon Sep 17 00:00:00 2001
From: Alan Liu <zaoxing@users.noreply.github.com>
Date: Fri, 18 Sep 2026 14:08:41 -0400
Subject: [PATCH 11/13] Judge a retry's charge against the same capacity
 decision as a stage

The retry and EEXIST-loser paths accounted a ready file this object had not
counted before without checking committed + reserved + n against max_bytes_.
With a 1000-byte reservation paused and a 1000-byte ready file created by a
second Spool object, the retry was admitted under a 1500 cap and both
stages completed at 2000. account_existing now runs the same decision the
fresh path does -- including the reconciliation scan -- and returns kFull
when the charge does not fit. The scan may count the retried path itself,
so the re-check charges n only when the path is not already in the rebuilt
ledger.

Remove has the matching race: another Spool object can delete the file
between this one's exists()/size checks and its unlink(). Both ENOENT
landings now unaccount the path (when this object was charging it) exactly
like the already-missing branch, instead of leaking bytes for a file that
is gone.

The C++ suite gains the retry-concurrent-with-a-reservation regression: it
fails against the pre-fix spool with 'a concurrent retry was admitted: ok'.
---
 native/csrc/store/spool.cpp              | 135 +++++++++++++++--------
 native/csrc/store/spool.h                |   7 ++
 tests/native/test_spool_reservations.cpp |  56 ++++++++++
 3 files changed, 151 insertions(+), 47 deletions(-)

diff --git a/native/csrc/store/spool.cpp b/native/csrc/store/spool.cpp
index 7027f84a0..d343c3a59 100644
--- a/native/csrc/store/spool.cpp
+++ b/native/csrc/store/spool.cpp
@@ -309,6 +309,49 @@ bool Spool::UnaccountReadyLocked(const std::string& path) {
   return true;
 }
 
+void Spool::ReconcileCommittedLocked() {
+  // The committed counter is only as fresh as the Remove calls THIS Spool
+  // object has seen. An uploader running through a second Spool object on
+  // the same root (or another process) removes ready files and updates its
+  // OWN counter -- this object's never learns about the released capacity,
+  // so staging eventually refuses on a directory that is actually empty
+  // (reproduced: sink with a 1500-byte limit, serial upload-and-remove
+  // between two records). Reconcile the COMMITTED account from the
+  // directory -- the durable truth for what is committed.
+  //
+  // The scan is authoritative for committed files only. The reservations
+  // other stagers hold right now are not on disk (or are, as a temp file
+  // this object already accounts for), so they are kept as they are and
+  // added back on top: replacing a single combined counter with the scan
+  // admitted a concurrent stage the reservation should have refused.
+  uint64_t actual = 0;
+  uint64_t ready_count = 0;
+  std::unordered_map<std::string, uint64_t> seen_ready;
+  std::error_code walk_ec;
+  for (const auto& entry :
+       fs::recursive_directory_iterator(root_, walk_ec)) {
+    if (walk_ec) break;
+    if (!entry.is_regular_file()) continue;
+    const std::string name = entry.path().filename().string();
+    if (HasSuffix(name, kReadySuffix)) {
+      actual += entry.file_size();
+      ++ready_count;
+      seen_ready.emplace(entry.path().string(), entry.file_size());
+    } else if (HasSuffix(name, kOpenSuffix) &&
+               inflight_temps_.count(entry.path().string()) == 0) {
+      // Someone else's in-progress write (another process, or a stale
+      // leftover Recover() has not swept yet): counts, as it does at
+      // Open(). This object's own temps are reservations.
+      actual += entry.file_size();
+    }
+  }
+  committed_bytes_ = actual;
+  committed_entries_ = ready_count;
+  // The path ledger is rebuilt with the aggregate it describes, so the two
+  // never disagree about which files the committed account holds.
+  accounted_ready_ = std::move(seen_ready);
+}
+
 void Spool::SetStageHookForTesting(std::function<void()> hook) {
   std::lock_guard<std::mutex> lock(mutex_);
   stage_hook_for_testing_ = std::move(hook);
@@ -388,48 +431,7 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns,
         }
       }
       if (committed_bytes_ + reserved_bytes_ + n > max_bytes_) {
-        // The committed counter is only as fresh as the Remove calls THIS
-        // Spool object has seen. An uploader running through a second
-        // Spool object on the same root (or another process) removes
-        // ready files and updates its OWN counter — this object's never
-        // learns about the released capacity, so staging eventually
-        // refuses on a directory that is actually empty (reproduced: sink
-        // with a 1500-byte limit, serial upload-and-remove between two
-        // records). Reconcile the COMMITTED account from the directory —
-        // the durable truth for what is committed — before refusing.
-        //
-        // The scan is authoritative for committed files only. The
-        // reservations other stagers hold right now are not on disk (or
-        // are, as a temp file this object already accounts for), so they
-        // are kept as they are and added back on top: replacing a single
-        // combined counter with the scan admitted a concurrent stage the
-        // reservation should have refused.
-        uint64_t actual = 0;
-        uint64_t ready_count = 0;
-        std::unordered_map<std::string, uint64_t> seen_ready;
-        std::error_code walk_ec;
-        for (const auto& entry :
-             fs::recursive_directory_iterator(root_, walk_ec)) {
-          if (walk_ec) break;
-          if (!entry.is_regular_file()) continue;
-          const std::string name = entry.path().filename().string();
-          if (HasSuffix(name, kReadySuffix)) {
-            actual += entry.file_size();
-            ++ready_count;
-            seen_ready.emplace(entry.path().string(), entry.file_size());
-          } else if (HasSuffix(name, kOpenSuffix) &&
-                     inflight_temps_.count(entry.path().string()) == 0) {
-            // Someone else's in-progress write (another process, or a
-            // stale leftover Recover() has not swept yet): counts, as it
-            // does at Open(). This object's own temps are reservations.
-            actual += entry.file_size();
-          }
-        }
-        committed_bytes_ = actual;
-        committed_entries_ = ready_count;
-        // The path ledger is rebuilt with the aggregate it describes, so the
-        // two never disagree about which files the committed account holds.
-        accounted_ready_ = std::move(seen_ready);
+        ReconcileCommittedLocked();
         if (committed_bytes_ + reserved_bytes_ + n > max_bytes_) {
           if (error) {
             *error = "spool byte limit exceeded: " +
@@ -469,10 +471,32 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns,
   };
   // The retry and EEXIST-loser paths both end on a ready file that already
   // exists. Whether it costs anything depends on whether THIS object has
-  // counted that path before, which only the path ledger can answer.
-  auto account_existing = [&] {
+  // counted that path before, which only the path ledger can answer. A path
+  // it has NOT counted is a charge like any other stage and must pass the
+  // same capacity decision -- including the reconciliation scan -- rather
+  // than being added on top of an in-flight reservation: with a 1000-byte
+  // reservation paused and a 1000-byte ready file created by a second Spool
+  // object, admitting the retry put committed + reserved at 2000 under a
+  // 1500 cap and both stages completed.
+  auto account_existing = [&]() -> SpoolStatus {
     std::lock_guard<std::mutex> lock(mutex_);
+    if (accounted_ready_.count(ready) != 0) return SpoolStatus::kOk;
+    if (committed_bytes_ + reserved_bytes_ + n > max_bytes_) {
+      ReconcileCommittedLocked();
+      // The scan may have just counted `ready` itself; its bytes are then
+      // already inside committed and charging n again would double-count.
+      const uint64_t charge = accounted_ready_.count(ready) != 0 ? 0 : n;
+      const uint64_t after = committed_bytes_ + reserved_bytes_ + charge;
+      if (after > max_bytes_) {
+        if (error) {
+          *error = "spool byte limit exceeded: " + std::to_string(after) +
+                   " > " + std::to_string(max_bytes_);
+        }
+        return SpoolStatus::kFull;
+      }
+    }
     if (AccountReadyLocked(ready, n)) ++generation_;
+    return SpoolStatus::kOk;
   };
   auto fill_out = [&] {
     out->pack_id = pack_id;
@@ -517,7 +541,8 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns,
     //
     // Every successful retry still closes the durability window itself with
     // a fresh fsync chain.
-    account_existing();
+    const SpoolStatus accounted = account_existing();
+    if (accounted != SpoolStatus::kOk) return accounted;
     if (!FsyncChain(root_, dir, error)) return SpoolStatus::kIo;
     fill_out();
     return SpoolStatus::kOk;
@@ -551,7 +576,8 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns,
         if (error) *error = "spool contains different content: " + object_key;
         return SpoolStatus::kConflict;
       }
-      account_existing();
+      const SpoolStatus accounted = account_existing();
+      if (accounted != SpoolStatus::kOk) return accounted;
       // The winner may still be between link() and its own fsync: do not
       // acknowledge its dirent before independently making the chain
       // durable (mirrors the Python loser's fsync).
@@ -686,11 +712,26 @@ SpoolStatus Spool::Remove(const StagedPack& staged, std::string* error) {
         id != staged.pack_id || created != staged.created_at_ns ||
         records != staged.record_count || sum != staged.checksum ||
         fs::file_size(staged.path, ec) != staged.object_bytes) {
+      if (ec == std::errc::no_such_file_or_directory) {
+        // The file vanished between the exists() check and this one --
+        // another Spool object removed it. Uncount the path if THIS object
+        // was charging it, exactly like the already-missing branch, or the
+        // ledger keeps bytes for a file that is gone and a later retry
+        // treats a recreated path as already accounted.
+        if (UnaccountReadyLocked(staged.path)) ++generation_;
+        return SpoolStatus::kOk;
+      }
       if (error) *error = "staged pack identity changed before removal";
       return SpoolStatus::kIntegrity;
     }
     if (::unlink(staged.path.c_str()) != 0) {
-      if (errno == ENOENT) return SpoolStatus::kOk;
+      if (errno == ENOENT) {
+        // Same removal race, now between the checks above and the unlink:
+        // release this object's charge for the path that another Spool
+        // object just removed.
+        if (UnaccountReadyLocked(staged.path)) ++generation_;
+        return SpoolStatus::kOk;
+      }
       if (error) *error = "cannot remove staged pack: " + std::string(strerror(errno));
       return SpoolStatus::kIo;
     }
diff --git a/native/csrc/store/spool.h b/native/csrc/store/spool.h
index a56005c21..ad3558653 100644
--- a/native/csrc/store/spool.h
+++ b/native/csrc/store/spool.h
@@ -118,6 +118,13 @@ class Spool {
   // staged elsewhere then retried here left room for another 1000).
   bool AccountReadyLocked(const std::string& path, uint64_t object_bytes);
   bool UnaccountReadyLocked(const std::string& path);
+  // Rebuild the COMMITTED account (bytes, entries, path ledger) from the
+  // directory, leaving in-flight reservations alone. Run under `mutex_`
+  // before refusing a stage: this object's counter is only as fresh as the
+  // Remove calls it has seen, so a second Spool object's removals make it
+  // stale-high. The retry/EEXIST-loser paths run it too, so a charge that
+  // would exceed the cap is judged against the same durable truth.
+  void ReconcileCommittedLocked();
 
   std::string root_;
   uint64_t max_bytes_ = 0;
diff --git a/tests/native/test_spool_reservations.cpp b/tests/native/test_spool_reservations.cpp
index 4f9a6372e..8a67f0fb3 100644
--- a/tests/native/test_spool_reservations.cpp
+++ b/tests/native/test_spool_reservations.cpp
@@ -12,6 +12,9 @@
 //      Open() is reached by this one's retry or EEXIST-loser path. This
 //      object's committed account has never counted it, so it has to count
 //      it now -- but exactly once, however many times that path is met.
+//   4. That same retry charge arrives while this object holds an in-flight
+//      reservation. It is a charge like any other and must pass the
+//      capacity decision, not be added on top of the reservation.
 //
 // A single counter cannot satisfy (1) and (2) at once: replacing it with
 // the scan fixes (1) and breaks (2) -- with max 1500, A reserved 1000, B's
@@ -328,6 +331,58 @@ void TestARetryOfAnotherObjectsReadyFileIsAccounted() {
   CHECK(writer.Snapshot().entries == 1);
 }
 
+// (4) A retry while THIS object holds an in-flight reservation. The retry
+// path used to charge a foreign ready file without passing the capacity
+// decision, so a paused reservation and a retry could both be admitted past
+// max_bytes_ (with a 1000-byte reservation and a 1000-byte foreign file
+// under a 1500 cap, both completed at 2000). The retry must now be refused
+// like any other charge that does not fit.
+void TestARetryCannotOversubscribeAnInflightReservation() {
+  const std::string root = FreshRoot("retry-vs-reservation");
+  dmi_store::SpoolConfig config{root, 1500};
+  dmi_store::Spool spool, other;
+  std::string error;
+  CHECK(dmi_store::Spool::Open(config, &spool, &error) ==
+        dmi_store::SpoolStatus::kOk);
+  CHECK(dmi_store::Spool::Open(config, &other, &error) ==
+        dmi_store::SpoolStatus::kOk);
+
+  dmi_store::SpoolStatus retry_status = dmi_store::SpoolStatus::kIo;
+  std::string retry_error;
+  spool.SetStageHookForTesting([&] {
+    // A's 1000-byte reservation is held and nothing of A's is on disk. A
+    // second Spool object stages a 1000-byte pack; the retry of that pack
+    // must be refused: committed + reserved + 1000 = 2000 > 1500.
+    dmi_store::StagedPack foreign;
+    std::string inner;
+    CHECK(StageBytes(other, 2, 1000, &foreign, &inner) ==
+          dmi_store::SpoolStatus::kOk);
+    dmi_store::StagedPack retried;
+    retry_status = StageBytes(spool, 2, 1000, &retried, &retry_error);
+  });
+  dmi_store::StagedPack out;
+  CHECK(StageBytes(spool, 1, 1000, &out, &error) ==
+        dmi_store::SpoolStatus::kOk);
+  spool.SetStageHookForTesting(nullptr);
+
+  CHECK(retry_status == dmi_store::SpoolStatus::kFull);
+  if (retry_status != dmi_store::SpoolStatus::kFull) {
+    std::cerr << "a concurrent retry was admitted: "
+              << dmi_store::SpoolStatusName(retry_status) << " "
+              << retry_error << "\n";
+  }
+  // What is refused is acknowledging the foreign pack ON TOP of the
+  // outstanding reservation. The capacity decision's reconciliation scan
+  // then records the foreign file anyway -- it IS on disk, and no object can
+  // stop the second writer -- so once A commits the account holds both, at
+  // the physical truth. 2000 bytes on disk against a 1500 cap is the second
+  // writer's doing, not this object admitting a stage it should have
+  // refused.
+  CHECK(spool.Snapshot().bytes == 2000);
+  CHECK(spool.Snapshot().entries == 2);
+  CHECK(BytesOnDisk(root) == 2000);
+}
+
 // The EEXIST-loser path reaches the same file by a different route: both
 // objects write the same pack, and the one whose link() loses must still
 // end up with those bytes in its account.
@@ -394,6 +449,7 @@ int main() {
   TestARefusedStageAndARetryLeaveTheAccountsAlone();
   TestALostLinkRaceReleasesItsReservation();
   TestARetryOfAnotherObjectsReadyFileIsAccounted();
+  TestARetryCannotOversubscribeAnInflightReservation();
   TestAnEexistLoserAccountsForTheWinnersFile();
   TestRepeatedRemovalDoesNotReleaseAnotherPacksCapacity();
   if (g_failures != 0) {

From 9823a61d77a41a2bfa2350c2ded468d06eea36e3 Mon Sep 17 00:00:00 2001
From: Alan Liu <zaoxing@users.noreply.github.com>
Date: Fri, 18 Sep 2026 14:08:44 -0400
Subject: [PATCH 12/13] Name the escape failure the decoder actually reported

A malformed \uXXXX or an unpaired surrogate is replaced with U+FFFD, which
IS valid UTF-8, so 'capture metadata text is not encodable UTF-8' named the
wrong failure; those two are the only ways the decoder clears its ok latch.
Both the production row path and the conformance driver now say what
happened: 'capture metadata text has an invalid or unpaired Unicode escape'.

Also fixes the footer test's 'bar' -> 'barring' wording.
---
 native/csrc/sink/conformance_sink.cpp        | 6 +++++-
 native/csrc/sink/record_row.cpp              | 6 +++++-
 tests/test_native_hydration_footer_fields.py | 2 +-
 3 files changed, 11 insertions(+), 3 deletions(-)

diff --git a/native/csrc/sink/conformance_sink.cpp b/native/csrc/sink/conformance_sink.cpp
index ada65833b..47196a0ff 100644
--- a/native/csrc/sink/conformance_sink.cpp
+++ b/native/csrc/sink/conformance_sink.cpp
@@ -179,7 +179,11 @@ int main() {
   };
   const auto refuse_bad_text = [] {
     std::string out = "{\"ok\":false,\"what\":";
-    jc::EscapeJson("capture metadata text is not encodable UTF-8", &out);
+    // 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)) {
diff --git a/native/csrc/sink/record_row.cpp b/native/csrc/sink/record_row.cpp
index b811d5fc5..932e0332c 100644
--- a/native/csrc/sink/record_row.cpp
+++ b/native/csrc/sink/record_row.cpp
@@ -67,7 +67,11 @@ bool ParseMetadataJson(const std::string& text,
   out->capture_policy_version = text_field("capture_policy_version");
   out->hook_name = text_field("hook_name");
   if (!text_ok) {
-    return fail("capture metadata text is not encodable UTF-8");
+    // 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.
diff --git a/tests/test_native_hydration_footer_fields.py b/tests/test_native_hydration_footer_fields.py
index cea463856..ac3969220 100644
--- a/tests/test_native_hydration_footer_fields.py
+++ b/tests/test_native_hydration_footer_fields.py
@@ -2,7 +2,7 @@
 
 hydrate binds every catalog descriptor to the pack footer by comparing the
 fields of the footer's rendered VALUES row -- one per CAPTURE_COLUMNS entry
-bar index_version (pack_index's renderer, SQL
+barring index_version (pack_index's renderer, SQL
 escaping and all) -- against the catalog row the reader returns (TSV escapes
 already undone, a NULL arriving as the empty string). The two sides only
 compare correctly in ONE representation, so the decoder is pinned here on

From 55470c84347dc1a3e3b1997c637ad8f197f8d2aa Mon Sep 17 00:00:00 2001
From: Alan Liu <zaoxing@users.noreply.github.com>
Date: Fri, 18 Sep 2026 14:26:54 -0400
Subject: [PATCH 13/13] Make the retry verdict independent of which call ran
 the scan

account_existing returned kOk unconditionally for a path already in the
ledger, but the first retry of a foreign file is refused only AFTER its own
reconciliation scan ledgers that path -- so an identical second retry
flipped from kFull to kOk with no state change but the scan's. The
already-counted branch now re-checks the same committed + reserved sum
instead of short-circuiting, and the regression drives the retry twice and
asserts the same verdict both times (fails on the previous commit with 'an
identical retry flipped to ok').

Also fold the reconciled committed total into peak_bytes_, which otherwise
could read below the bytes the account holds right now.
---
 native/csrc/store/spool.cpp              | 43 +++++++++++++-----------
 tests/native/test_spool_reservations.cpp | 11 ++++++
 2 files changed, 35 insertions(+), 19 deletions(-)

diff --git a/native/csrc/store/spool.cpp b/native/csrc/store/spool.cpp
index d343c3a59..3dab9f242 100644
--- a/native/csrc/store/spool.cpp
+++ b/native/csrc/store/spool.cpp
@@ -350,6 +350,9 @@ void Spool::ReconcileCommittedLocked() {
   // The path ledger is rebuilt with the aggregate it describes, so the two
   // never disagree about which files the committed account holds.
   accounted_ready_ = std::move(seen_ready);
+  // The scan can raise the committed total (files another object wrote), and
+  // peak_bytes_ must never read below what the account holds right now.
+  peak_bytes_ = std::max(peak_bytes_, committed_bytes_ + reserved_bytes_);
 }
 
 void Spool::SetStageHookForTesting(std::function<void()> hook) {
@@ -472,28 +475,30 @@ SpoolStatus Spool::Stage(const std::string& pack_id, uint64_t created_at_ns,
   // The retry and EEXIST-loser paths both end on a ready file that already
   // exists. Whether it costs anything depends on whether THIS object has
   // counted that path before, which only the path ledger can answer. A path
-  // it has NOT counted is a charge like any other stage and must pass the
-  // same capacity decision -- including the reconciliation scan -- rather
-  // than being added on top of an in-flight reservation: with a 1000-byte
-  // reservation paused and a 1000-byte ready file created by a second Spool
-  // object, admitting the retry put committed + reserved at 2000 under a
-  // 1500 cap and both stages completed.
+  // it has NOT counted is judged against the durable truth plus in-flight
+  // reservations -- the same capacity decision a fresh stage makes, except
+  // that the reconciliation scan counts the file's bytes rather than
+  // charging them on top of the committed total. Without the decision, a
+  // retry could be added on top of an in-flight reservation: with a
+  // 1000-byte reservation paused and a 1000-byte ready file created by a
+  // second Spool object, admitting the retry put committed + reserved at
+  // 2000 under a 1500 cap and both stages completed.
+  //
+  // The verdict must not depend on which call ran the scan. An
+  // already-counted path skips the scan, so it re-checks the same sum
+  // instead of returning kOk unconditionally: otherwise the first retry
+  // (which reconciles, sees the overage and refuses) and the second (which
+  // finds the path ledgered) would disagree about identical state.
   auto account_existing = [&]() -> SpoolStatus {
     std::lock_guard<std::mutex> lock(mutex_);
-    if (accounted_ready_.count(ready) != 0) return SpoolStatus::kOk;
-    if (committed_bytes_ + reserved_bytes_ + n > max_bytes_) {
-      ReconcileCommittedLocked();
-      // The scan may have just counted `ready` itself; its bytes are then
-      // already inside committed and charging n again would double-count.
-      const uint64_t charge = accounted_ready_.count(ready) != 0 ? 0 : n;
-      const uint64_t after = committed_bytes_ + reserved_bytes_ + charge;
-      if (after > max_bytes_) {
-        if (error) {
-          *error = "spool byte limit exceeded: " + std::to_string(after) +
-                   " > " + std::to_string(max_bytes_);
-        }
-        return SpoolStatus::kFull;
+    if (accounted_ready_.count(ready) == 0) ReconcileCommittedLocked();
+    if (committed_bytes_ + reserved_bytes_ > max_bytes_) {
+      if (error) {
+        *error = "spool byte limit exceeded: " +
+                 std::to_string(committed_bytes_ + reserved_bytes_) + " > " +
+                 std::to_string(max_bytes_);
       }
+      return SpoolStatus::kFull;
     }
     if (AccountReadyLocked(ready, n)) ++generation_;
     return SpoolStatus::kOk;
diff --git a/tests/native/test_spool_reservations.cpp b/tests/native/test_spool_reservations.cpp
index 8a67f0fb3..8acd5cba4 100644
--- a/tests/native/test_spool_reservations.cpp
+++ b/tests/native/test_spool_reservations.cpp
@@ -348,6 +348,7 @@ void TestARetryCannotOversubscribeAnInflightReservation() {
         dmi_store::SpoolStatus::kOk);
 
   dmi_store::SpoolStatus retry_status = dmi_store::SpoolStatus::kIo;
+  dmi_store::SpoolStatus second_retry_status = dmi_store::SpoolStatus::kIo;
   std::string retry_error;
   spool.SetStageHookForTesting([&] {
     // A's 1000-byte reservation is held and nothing of A's is on disk. A
@@ -359,6 +360,10 @@ void TestARetryCannotOversubscribeAnInflightReservation() {
           dmi_store::SpoolStatus::kOk);
     dmi_store::StagedPack retried;
     retry_status = StageBytes(spool, 2, 1000, &retried, &retry_error);
+    // And the SAME call must answer the same way after its own
+    // reconciliation scan ledgered the foreign path: an identical retry
+    // cannot flip from kFull to kOk without any state change but its own.
+    second_retry_status = StageBytes(spool, 2, 1000, &retried, &retry_error);
   });
   dmi_store::StagedPack out;
   CHECK(StageBytes(spool, 1, 1000, &out, &error) ==
@@ -371,6 +376,12 @@ void TestARetryCannotOversubscribeAnInflightReservation() {
               << dmi_store::SpoolStatusName(retry_status) << " "
               << retry_error << "\n";
   }
+  CHECK(second_retry_status == dmi_store::SpoolStatus::kFull);
+  if (second_retry_status != dmi_store::SpoolStatus::kFull) {
+    std::cerr << "an identical retry flipped to "
+              << dmi_store::SpoolStatusName(second_retry_status) << " "
+              << retry_error << "\n";
+  }
   // What is refused is acknowledging the foreign pack ON TOP of the
   // outstanding reservation. The capacity decision's reconciliation scan
   // then records the foreign file anyway -- it IS on disk, and no object can