fix(autopipeline): locate BLOCK by position, not value

isBlockingCmd stopped its XREAD/XREADGROUP/TS.READ arg scan on the
first literal "streams" token, treating it as the structural STREAMS
keyword. But STREAMS is only a real terminator for XREAD/XREADGROUP,
and XREADGROUP's GROUP/consumer names (fixed-position, arbitrary
values) sit before the option section — a consumer literally named
"streams" hit that case first and returned non-blocking before ever
reaching a real, later BLOCK option. TS.READ has no STREAMS keyword
at all, so a key or timestamp named "streams" tripped the same case
for no reason. Either way a genuinely blocking raw command could ride
the shared full-duplex pipe and stall it (cursor bugbot on #4002).

Split the scan per command: XREAD starts after the command name,
XREADGROUP skips the fixed GROUP/consumer positions before scanning,
and TS.READ (which never has a STREAMS terminator) just looks for
BLOCK anywhere. Adds coverage for group/consumer/key names that
collide with "streams" or "block".

autopipeline.go

@@ -1741,20 +1741,48 @@ func isBlockingCmd(cmd Cmder) bool {
 	// Arg-driven: these block only in their BLOCK form, and blanket-diverting
 	// the far more common non-blocking form would drop it out of batching for
 	// nothing. TS.READ takes BLOCK the same way (see TSReadWithArgs).
-	if name != "xread" && name != "xreadgroup" && name != "ts.read" {
+	args := cmd.Args()
+	switch name {
+	case "xread":
+		return blockOptionBeforeStreams(args, 1)
+	case "xreadgroup":
+		// args[1:4] are the mandatory GROUP keyword plus the group and
+		// consumer names. Those names are arbitrary values — a consumer
+		// literally named "streams" must not be mistaken for the STREAMS
+		// terminator (which would hide a real, later BLOCK and let the
+		// command ride the shared pipe — cursor bugbot on #4002) — so skip
+		// them positionally rather than matching on value.
+		return blockOptionBeforeStreams(args, 4)
+	case "ts.read":
+		// TS.READ has no STREAMS terminator at all, so there is nothing to
+		// stop the scan early for: the key and timestamp are arbitrary
+		// values (e.g. a key literally named "streams") that must not be
+		// treated as one. Just look for the BLOCK keyword anywhere.
+		for _, arg := range args {
+			if internal.ToLower(blockingArgString(arg)) == "block" {
+				return true
+			}
+		}
+		return false
+	default:
 		return false
 	}
-	// Match the token the way the encoder does: a raw Cmder may carry RESP
-	// tokens as []byte or *string (see baseCmd.stringArg), and a type switch on
-	// string alone would let NewCmd(ctx, "xread", []byte("BLOCK"), 0, ...) be
-	// batched onto a shared connection.
-	// BLOCK is an OPTION that appears before the STREAMS keyword; the keys and IDs
-	// that follow STREAMS may be literally "block" (e.g.
-	// XReadArgs{Streams: []string{"block", "0"}}). Stop at STREAMS so a stream key
-	// is never mistaken for the option — that would divert a non-blocking
-	// XREAD/XREADGROUP off the ordered pipe and let a later command run first.
-	// ts.read carries no STREAMS token, so its scan is unchanged (whole-arg).
-	for _, arg := range cmd.Args() {
+}
+
+// blockOptionBeforeStreams scans the option section of XREAD/XREADGROUP
+// (COUNT/MAXCOUNT/MAXSIZE/BLOCK/NOACK/CLAIM, in any combination) for the
+// BLOCK keyword, stopping at the STREAMS keyword that always terminates the
+// option section. start must already be past any positional arguments
+// (XREADGROUP's GROUP clause) that cannot be told apart from a keyword by
+// value alone. Match the token the way the encoder does: a raw Cmder may
+// carry RESP tokens as []byte or *string (see baseCmd.stringArg), and a type
+// switch on string alone would let NewCmd(ctx, "xread", []byte("BLOCK"), 0,
+// ...) be batched onto a shared connection.
+func blockOptionBeforeStreams(args []interface{}, start int) bool {
+	if start > len(args) {
+		return false
+	}
+	for _, arg := range args[start:] {
 		switch internal.ToLower(blockingArgString(arg)) {
 		case "streams":
 			return false

autopipeline_internal_test.go

@@ -1534,6 +1534,52 @@ func TestBlockingDetectionStopsAtStreams(t *testing.T) {
 	}
 }
 
+// TestBlockingDetectionGroupConsumerNamedStreams pins that a value earlier in
+// the arg list literally equal to "streams" or "block" is never mistaken for
+// a keyword: XREADGROUP's GROUP/consumer names sit at fixed positions before
+// the option section, so a consumer or group named "streams" must not be
+// read as the STREAMS terminator (which would hide a later, real BLOCK and
+// let the command ride the shared pipe instead of being diverted — cursor
+// bugbot on #4002), and a group/consumer named "block" must not be read as
+// the BLOCK option (which would divert a non-blocking command for nothing).
+func TestBlockingDetectionGroupConsumerNamedStreams(t *testing.T) {
+	ctx := context.Background()
+	blocking := [][]interface{}{
+		{"xreadgroup", "GROUP", "streams", "c", "BLOCK", 100, "STREAMS", "s", ">"},
+		{"xreadgroup", "GROUP", "g", "streams", "BLOCK", 100, "STREAMS", "s", ">"},
+		{"xreadgroup", "GROUP", "block", "c", "BLOCK", 100, "STREAMS", "s", ">"},
+	}
+	for _, args := range blocking {
+		if !isBlockingCmd(NewCmd(ctx, args...)) {
+			t.Errorf("isBlockingCmd(%v) = false, want true (BLOCK option present after group/consumer named streams/block)", args)
+		}
+	}
+	nonBlocking := [][]interface{}{
+		{"xreadgroup", "GROUP", "streams", "c", "STREAMS", "s", ">"},
+		{"xreadgroup", "GROUP", "block", "c", "STREAMS", "s", ">"},
+	}
+	for _, args := range nonBlocking {
+		if isBlockingCmd(NewCmd(ctx, args...)) {
+			t.Errorf("isBlockingCmd(%v) = true, want false (group/consumer named streams/block, no real BLOCK option)", args)
+		}
+	}
+}
+
+// TestBlockingDetectionTSReadKeyNamedStreams pins that TS.READ, which has no
+// STREAMS terminator at all, never stops its scan early on a key or
+// timestamp literally equal to "streams" — that would hide a later, real
+// BLOCK option and let the command ride the shared pipe (cursor bugbot on
+// #4002).
+func TestBlockingDetectionTSReadKeyNamedStreams(t *testing.T) {
+	ctx := context.Background()
+	if !isBlockingCmd(NewCmd(ctx, "ts.read", "streams", "-", "BLOCK", 100, 1)) {
+		t.Error(`isBlockingCmd(ts.read key="streams" ... BLOCK) = false, want true`)
+	}
+	if isBlockingCmd(NewCmd(ctx, "ts.read", "streams", "-", "MAX_COUNT", 10)) {
+		t.Error(`isBlockingCmd(ts.read key="streams", no BLOCK) = true, want false`)
+	}
+}
+
 // TestDivertRegistrationRacesClose hammers the window between a diverted
 // command's closed check and its registration against Close: every submitted
 // command must end with a definite outcome, and Close must never report