fix(redis,stream): XADD '*' avoids 0-0 when clock is pre-epoch

Gemini medium + Codex P2 round 1 on PR #631: the original change
clamped a pre-epoch clock to nowMs=0 via safeUnixMilliToUint64, but on
a fresh stream the auto-ID branch then returned "0-0" — the very ID
the same function rejects as invalid for explicitly-requested IDs and
which XREAD ... 0 treats as the empty after-marker.

Fix: extract the auto-ID branch into autoXAddID(nowMs, hasLast, ...)
so it can be unit-tested without the wall clock; bump seq to 1 when
nowMs is 0 so the first auto-generated entry is "0-1" rather than
"0-0". The existing bumpStreamID-on-collision path is unchanged.

Tests:

- TestAutoXAddID covers the fresh-stream branch (sane clock, clamped
  clock → 0-1), the same-ms collision (bump to seq+1), the
  clock-behind-lastMs case (carry from lastMs/lastSeq, not nowMs),
  the seq-at-MaxUint64 carry to ms+1, and the ID-space-exhausted
  error.

Build / vet / lint clean.

Author: bootjp

adapter/redis_compat_commands.go

@@ -3702,10 +3702,34 @@ func nextXAddID(hasLast bool, lastMs, lastSeq uint64, requested string) (string,
 		}
 		return requested, nil
 	}
+	return autoXAddID(safeUnixMilliToUint64(time.Now().UnixMilli()), hasLast, lastMs, lastSeq)
+}
 
-	nowMs := safeUnixMilliToUint64(time.Now().UnixMilli())
+// autoXAddID resolves XADD '*' to a concrete stream ID given a wall-clock
+// nowMs. Pulled out of nextXAddID so the auto-ID branch is testable
+// without depending on time.Now() — the only un-injectable dependency is
+// already isolated in the caller.
+//
+// Two corner cases the caller cannot rely on the wall clock to avoid:
+//
+//   - nowMs == 0 on a fresh stream (!hasLast). A naive "<nowMs>-0" reply
+//     yields "0-0", which Redis explicitly rejects as a stream ID and
+//     which XREAD ... 0 would treat as the empty after-marker. Bump the
+//     seq to 1 so the first auto-generated entry is "0-1" — strictly
+//     greater than 0-0 and reachable via XREAD ... 0. (This case fires
+//     only when safeUnixMilliToUint64 clamped a pre-epoch clock to 0;
+//     under any sane clock, nowMs is well above 0.)
+//
+//   - nowMs <= lastMs. Advance past lastMs/lastSeq via bumpStreamID so
+//     the stream stays strictly monotonic even across a backwards clock
+//     step or a corrupted meta where lastMs is far in the future.
+func autoXAddID(nowMs uint64, hasLast bool, lastMs, lastSeq uint64) (string, error) {
 	if !hasLast || nowMs > lastMs {
-		return strconv.FormatUint(nowMs, 10) + "-0", nil
+		seq := uint64(0)
+		if nowMs == 0 {
+			seq = 1
+		}
+		return strconv.FormatUint(nowMs, 10) + "-" + strconv.FormatUint(seq, 10), nil
 	}
 	// Either nowMs == lastMs (same millisecond), or lastMs is in the future
 	// (monotonic guarantee across a backwards clock step or a corrupted

adapter/redis_stream_limit_test.go

@@ -338,3 +338,58 @@ func TestSafeUnixMilliToUint64(t *testing.T) {
 		})
 	}
 }
+
+// TestAutoXAddID covers the XADD '*' path of nextXAddID with synthetic
+// nowMs values, including the Codex P2 / Gemini-medium edge case:
+// safeUnixMilliToUint64 clamps a pre-epoch clock to 0, and a naive
+// "0-0" auto-ID is rejected by Redis (XREAD ... 0 treats it as the
+// after-marker and skips it). autoXAddID must bump seq to 1 in that
+// case so the first auto-generated entry is "0-1".
+func TestAutoXAddID(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name    string
+		nowMs   uint64
+		hasLast bool
+		lastMs  uint64
+		lastSeq uint64
+		want    string
+		wantErr bool
+	}{
+		// Fresh stream, healthy clock: nowMs > 0, seq starts at 0.
+		{"fresh stream, sane clock", 1_777_000_000_000, false, 0, 0, "1777000000000-0", false},
+		// Fresh stream, clock pre-epoch (clamped to 0): MUST yield 0-1
+		// rather than 0-0 — the original Codex P2 / Gemini-medium case.
+		{"fresh stream, clamped clock → 0-1", 0, false, 0, 0, "0-1", false},
+		// Existing stream, nowMs strictly greater: seq resets to 0.
+		{"clock advanced past lastMs", 200, true, 100, 5, "200-0", false},
+		// Existing stream, nowMs == lastMs: bumpStreamID seq carry.
+		{"same ms as lastMs", 100, true, 100, 5, "100-6", false},
+		// Existing stream, nowMs < lastMs (clock went backwards):
+		// bumpStreamID carries from lastMs/lastSeq, NOT from nowMs.
+		{"clock behind lastMs", 50, true, 100, 5, "100-6", false},
+		// seq at MaxUint64 carries to ms+1.
+		{"seq at max carries", 100, true, 100, ^uint64(0), "101-0", false},
+		// Both ms and seq at MaxUint64: ID space exhausted, error.
+		{"ID space exhausted", 100, true, ^uint64(0), ^uint64(0), "", true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			got, err := autoXAddID(tc.nowMs, tc.hasLast, tc.lastMs, tc.lastSeq)
+			if tc.wantErr {
+				if err == nil {
+					t.Fatalf("autoXAddID(%d,%v,%d,%d): expected error, got %q", tc.nowMs, tc.hasLast, tc.lastMs, tc.lastSeq, got)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("autoXAddID(%d,%v,%d,%d): unexpected error %v", tc.nowMs, tc.hasLast, tc.lastMs, tc.lastSeq, err)
+			}
+			if got != tc.want {
+				t.Fatalf("autoXAddID(%d,%v,%d,%d): want %q, got %q", tc.nowMs, tc.hasLast, tc.lastMs, tc.lastSeq, tc.want, got)
+			}
+		})
+	}
+}