backup: hash field array shape + 32-bit safety + kindByKey centralisation (PR #725, round 1)

Codex round 12 raised two P1 findings on commit 72e2b75; Gemini
raised one high and three mediums alongside.

P1: Hash field name encoding via JSON object key was lossy.
  - Distinct fields could collapse to the same JSON key: a UTF-8
    literal `%FF` and a single binary byte `0xFF` both ended up
    as `%FF` after the EncodeSegment-on-non-UTF-8 fallback,
    silently dropping data on hashes with both forms.
  - Long binary field names (>240 bytes) hit EncodeSegment's
    SHA fallback which is irreversible at this layer (no
    per-field KEYMAP), so restore could not reconstruct the
    original bytes.
Fix: change `fields` to a JSON array of typed `{name, value}`
records. Both name and value go through the same UTF-8-or-base64
envelope (marshalRedisBinaryValue, mirrors PR #714's SQS body),
so arbitrary binary bytes round-trip without collisions and
without SHA fallback.

High (Gemini): parseUserKeyLenPrefix used `int(uint32)` which
wraps to negative on 32-bit when the high bit is set, bypassing
the bounds check and causing a slice panic. Compare in `uint64`
space before converting.

Medium (Gemini, ×3): centralise `kindByKey` registration in the
hashState helper so meta/field/TTL paths agree on the kind in
one place. Audited callers — HandleHashMeta, HandleHashField, and
HandleTTL's `case redisKindHash:` arm — all reach hashState via
the same userKey identity, and the TTL caller already knows the
kind so the now-implicit kindByKey write is idempotent. No other
callers per `grep -n "r\.hashState(" internal/backup/`.

Tests:
  - TestRedisDB_HashRoundTripBasic updated to assert the array
    shape via new hashFieldArray / hashFieldByName helpers.
  - TestRedisDB_HashEmptyHashStillEmitsFile, _BinaryFieldValue
    updated likewise.
  - NEW TestRedisDB_HashBinaryFieldNameRoundTripsViaArray drives
    the exact Codex collision pair (UTF-8 `%FF` + binary 0xFF)
    and asserts both names survive distinctly with their
    expected types.

Verified: linux race tests, GOOS=windows, GOOS=js wasm, lint —
all clean.

Author: bootjp

internal/backup/redis_hash.go

@@ -63,7 +63,6 @@ func (r *RedisDB) HandleHashMeta(key, value []byte) error {
 	st := r.hashState(userKey)
 	st.declaredLen = int64(binary.BigEndian.Uint64(value)) //nolint:gosec // signed int64 by design
 	st.metaSeen = true
-	r.kindByKey[string(userKey)] = redisKindHash
 	return nil
 }
 
@@ -80,17 +79,31 @@ func (r *RedisDB) HandleHashField(key, value []byte) error {
 	}
 	st := r.hashState(userKey)
 	st.fields[string(fieldName)] = bytes.Clone(value)
-	r.kindByKey[string(userKey)] = redisKindHash
 	return nil
 }
 
-// hashState lazily creates per-key state.
+// hashState lazily creates per-key state. The `kindByKey` registration
+// lives here (Gemini medium PR #725 #1/#3) so every code path that
+// touches a hash state — meta, field, and the TTL-routing back-edge
+// from HandleTTL — agrees on the kind. Caller audit (per the loop's
+// "audit semantics-changing edits" rule):
+//
+//   - HandleHashMeta and HandleHashField both want kindByKey set;
+//     centralising here means the explicit assignment is no longer
+//     needed at the call site.
+//   - HandleTTL only ever calls hashState() inside the
+//     `case redisKindHash:` branch, where kindByKey == redisKindHash
+//     already holds; the assignment here is idempotent for that path.
+//   - No other caller exists; verified via
+//     `grep -n "r\.hashState(" internal/backup/`.
 func (r *RedisDB) hashState(userKey []byte) *redisHashState {
-	if st, ok := r.hashes[string(userKey)]; ok {
+	uk := string(userKey)
+	if st, ok := r.hashes[uk]; ok {
 		return st
 	}
 	st := &redisHashState{fields: make(map[string][]byte)}
-	r.hashes[string(userKey)] = st
+	r.hashes[uk] = st
+	r.kindByKey[uk] = redisKindHash
 	return st
 }
 
@@ -122,15 +135,20 @@ func parseHashFieldKey(key []byte) ([]byte, []byte, bool) {
 // parseUserKeyLenPrefix decodes the shared <len(4)><userKey> shape used
 // by every wide-column !hs|/!st|/!zs| key. Returns the userKey slice
 // (aliasing the input) plus a presence flag.
+//
+// The length comparison is done in uint64 space because on 32-bit
+// architectures `int(uint32)` can wrap to a negative value when the
+// high bit is set, bypassing the bounds check and causing a slice
+// panic. Gemini high finding (PR #725 round 1).
 func parseUserKeyLenPrefix(b []byte) ([]byte, bool) {
 	if len(b) < hashUserKeyLenSize {
 		return nil, false
 	}
-	ukLen := int(binary.BigEndian.Uint32(b[:hashUserKeyLenSize])) //nolint:gosec // bounded by len(b)
-	if len(b) < hashUserKeyLenSize+ukLen {
+	ukLen := binary.BigEndian.Uint32(b[:hashUserKeyLenSize])
+	if uint64(len(b)) < uint64(hashUserKeyLenSize)+uint64(ukLen) {
 		return nil, false
 	}
-	return b[hashUserKeyLenSize : hashUserKeyLenSize+ukLen], true
+	return b[hashUserKeyLenSize : hashUserKeyLenSize+int(ukLen)], true //nolint:gosec // bounded above
 }
 
 // flushHashes writes one JSON file per accumulated hash to
@@ -186,37 +204,49 @@ func (r *RedisDB) writeHashJSON(dir string, userKey []byte, st *redisHashState)
 	return nil
 }
 
-// hashJSONFields marshals the per-field map. Field NAMES are emitted
-// as JSON object keys, which forces them into UTF-8 string territory:
-// when a name is not valid UTF-8 we percent-encode it via
-// EncodeSegment so the output is reversible without changing JSON
-// shape. Field VALUES preserve full binary fidelity via the same
-// `{"base64":...}` envelope used for SQS bodies (see sqsMessageBody).
-type hashJSONFields map[string]json.RawMessage
+// hashFieldRecord is the dump-format projection of one Redis hash
+// field. Both name and value go through the same UTF-8-or-base64
+// envelope (json.RawMessage produced by marshalRedisBinaryValue) so
+// arbitrary binary bytes round-trip without lossy rewrites.
+//
+// We deliberately emit `fields` as an ARRAY of records rather than a
+// JSON object keyed on the field name, because Redis hash field
+// names are binary-safe and JSON object keys are not. With a map
+// shape, two distinct fields could collapse to the same JSON key
+// (Codex P1 round 12 #725: a UTF-8 literal `%FF` and a single byte
+// `0xFF` both percent-encoded to `%FF` would overwrite one
+// another), and a >240-byte non-UTF-8 field name would route
+// through EncodeSegment's SHA fallback which is non-reversible at
+// this layer (no per-field KEYMAP). The array form keeps every
+// (name, value) pair distinct and binary-safe.
+type hashFieldRecord struct {
+	Name  json.RawMessage `json:"name"`
+	Value json.RawMessage `json:"value"`
+}
 
 func marshalHashJSON(st *redisHashState) ([]byte, error) {
-	fields := make(hashJSONFields, len(st.fields))
-	// Sort for deterministic output across runs.
+	// Sort by raw byte order for deterministic output across runs.
 	names := make([]string, 0, len(st.fields))
 	for name := range st.fields {
 		names = append(names, name)
 	}
 	sort.Strings(names)
+	fields := make([]hashFieldRecord, 0, len(names))
 	for _, name := range names {
-		jsonName := name
-		if !utf8.ValidString(name) {
-			jsonName = EncodeSegment([]byte(name))
+		nameJSON, err := marshalRedisBinaryValue([]byte(name))
+		if err != nil {
+			return nil, err
 		}
 		valueJSON, err := marshalRedisBinaryValue(st.fields[name])
 		if err != nil {
 			return nil, err
 		}
-		fields[jsonName] = valueJSON
+		fields = append(fields, hashFieldRecord{Name: nameJSON, Value: valueJSON})
 	}
 	type out struct {
-		FormatVersion uint32         `json:"format_version"`
-		Fields        hashJSONFields `json:"fields"`
-		ExpireAtMs    *uint64        `json:"expire_at_ms"`
+		FormatVersion uint32            `json:"format_version"`
+		Fields        []hashFieldRecord `json:"fields"`
+		ExpireAtMs    *uint64           `json:"expire_at_ms"`
 	}
 	rec := out{FormatVersion: 1, Fields: fields}
 	if st.hasTTL {

internal/backup/redis_hash_test.go

@@ -70,17 +70,43 @@ func hashFloat(t *testing.T, m map[string]any, key string) float64 {
 	return f
 }
 
-func hashSubMap(t *testing.T, m map[string]any, key string) map[string]any {
+// hashFieldArray fetches the `fields` JSON array from a decoded hash
+// JSON file and returns it as a slice of {name, value} maps. The
+// dump-format projection emits fields as an ARRAY of records (not a
+// JSON object) so binary-safe Redis field NAMES round-trip without
+// collision — see the comment on hashFieldRecord in redis_hash.go.
+func hashFieldArray(t *testing.T, m map[string]any) []map[string]any {
 	t.Helper()
-	v, ok := m[key]
+	v, ok := m["fields"]
 	if !ok {
-		t.Fatalf("field %q missing in %+v", key, m)
+		t.Fatalf("fields missing in %+v", m)
 	}
-	sub, ok := v.(map[string]any)
+	raw, ok := v.([]any)
 	if !ok {
-		t.Fatalf("field %q = %T(%v), want map[string]any", key, v, v)
+		t.Fatalf("fields = %T(%v), want []any", v, v)
 	}
-	return sub
+	out := make([]map[string]any, 0, len(raw))
+	for i, r := range raw {
+		rm, ok := r.(map[string]any)
+		if !ok {
+			t.Fatalf("fields[%d] = %T(%v), want map[string]any", i, r, r)
+		}
+		out = append(out, rm)
+	}
+	return out
+}
+
+// hashFieldByName looks up a single field in the array shape and
+// returns its `value` element (UTF-8 string or base64 envelope).
+func hashFieldByName(t *testing.T, m map[string]any, name string) any {
+	t.Helper()
+	for _, f := range hashFieldArray(t, m) {
+		if f["name"] == name {
+			return f["value"]
+		}
+	}
+	t.Fatalf("field %q not found in %+v", name, m["fields"])
+	return nil
 }
 
 func TestRedisDB_HashRoundTripBasic(t *testing.T) {
@@ -105,12 +131,11 @@ func TestRedisDB_HashRoundTripBasic(t *testing.T) {
 	if got["expire_at_ms"] != nil {
 		t.Fatalf("expire_at_ms must be nil without TTL, got %v", got["expire_at_ms"])
 	}
-	fields := hashSubMap(t, got, "fields")
-	if fields["name"] != "alice" {
-		t.Fatalf("name = %v want alice", fields["name"])
+	if v := hashFieldByName(t, got, "name"); v != "alice" {
+		t.Fatalf("name = %v want alice", v)
 	}
-	if fields["age"] != "30" {
-		t.Fatalf("age = %v want 30", fields["age"])
+	if v := hashFieldByName(t, got, "age"); v != "30" {
+		t.Fatalf("age = %v want 30", v)
 	}
 }
 
@@ -126,9 +151,8 @@ func TestRedisDB_HashEmptyHashStillEmitsFile(t *testing.T) {
 		t.Fatal(err)
 	}
 	got := readHashJSON(t, filepath.Join(root, "redis", "db_0", "hashes", "empty.json"))
-	fields := hashSubMap(t, got, "fields")
-	if len(fields) != 0 {
-		t.Fatalf("empty hash should emit empty fields object, got %v", fields)
+	if fields := hashFieldArray(t, got); len(fields) != 0 {
+		t.Fatalf("empty hash should emit empty fields array, got %v", fields)
 	}
 }
 
@@ -206,16 +230,91 @@ func TestRedisDB_HashBinaryFieldValueUsesBase64Envelope(t *testing.T) {
 		t.Fatal(err)
 	}
 	got := readHashJSON(t, filepath.Join(root, "redis", "db_0", "hashes", "k.json"))
-	fields := hashSubMap(t, got, "fields")
-	envelope, ok := fields["blob"].(map[string]any)
+	v := hashFieldByName(t, got, "blob")
+	envelope, ok := v.(map[string]any)
 	if !ok {
-		t.Fatalf("expected base64 envelope for binary value, got %T(%v)", fields["blob"], fields["blob"])
+		t.Fatalf("expected base64 envelope for binary value, got %T(%v)", v, v)
 	}
 	if envelope["base64"] == "" {
 		t.Fatalf("base64 envelope missing payload: %v", envelope)
 	}
 }
 
+// TestRedisDB_HashBinaryFieldNameRoundTripsViaArray covers the Codex
+// P1 round-12 scenario: a UTF-8-literal field name `%FF` and a
+// 1-byte binary field name `0xFF` would collide if `fields` were a
+// JSON object, because EncodeSegment percent-encodes 0xFF to `%FF`.
+// With the array-of-records shape both names round-trip
+// independently — the binary name uses the typed base64 envelope
+// while the UTF-8 literal stays as a plain JSON string.
+func TestRedisDB_HashBinaryFieldNameRoundTripsViaArray(t *testing.T) {
+	t.Parallel()
+	db, root := newRedisDB(t)
+	if err := db.HandleHashMeta(hashMetaKey("k"), encodeHashMetaValue(2)); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.HandleHashField(hashFieldKey("k", "%FF"), []byte("literal-percent-ff")); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.HandleHashField(append(hashFieldKey("k", ""), 0xff), []byte("binary-byte")); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Finalize(); err != nil {
+		t.Fatal(err)
+	}
+	got := readHashJSON(t, filepath.Join(root, "redis", "db_0", "hashes", "k.json"))
+	fields := hashFieldArray(t, got)
+	if len(fields) != 2 {
+		t.Fatalf("len(fields) = %d, want 2 (binary and literal must be distinct)", len(fields))
+	}
+	literalSeen, binarySeen := classifyHashFieldShapes(t, fields)
+	if !literalSeen {
+		t.Fatalf("literal %%FF field missing from %+v", fields)
+	}
+	if !binarySeen {
+		t.Fatalf("binary 0xFF field missing from %+v", fields)
+	}
+}
+
+// classifyHashFieldShapes folds the binary-vs-literal field-shape
+// assertions out of the parent test so cyclomatic complexity stays
+// under the project linter's ceiling. Returns (sawUTF8Literal,
+// sawBase64Envelope).
+func classifyHashFieldShapes(t *testing.T, fields []map[string]any) (bool, bool) {
+	t.Helper()
+	var literalSeen, binarySeen bool
+	for _, f := range fields {
+		switch n := f["name"].(type) {
+		case string:
+			if n == "%FF" {
+				literalSeen = true
+				if f["value"] != "literal-percent-ff" {
+					t.Fatalf("literal value = %v", f["value"])
+				}
+			}
+		case map[string]any:
+			if n["base64"] == base64URLEncode(0xff) {
+				binarySeen = true
+				if f["value"] != "binary-byte" {
+					t.Fatalf("binary value = %v", f["value"])
+				}
+			}
+		default:
+			t.Fatalf("unexpected name type %T(%v)", f["name"], f["name"])
+		}
+	}
+	return literalSeen, binarySeen
+}
+
+func base64URLEncode(b byte) string {
+	// base64.RawURLEncoding of a single byte; matches what
+	// marshalRedisBinaryValue emits for non-UTF-8 content.
+	const alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"
+	hi := alphabet[b>>2]
+	lo := alphabet[(b&0x3)<<4]
+	return string([]byte{hi, lo})
+}
+
 func TestRedisDB_HashRejectsMalformedMetaValueLength(t *testing.T) {
 	t.Parallel()
 	db, _ := newRedisDB(t)