test: Fix Redis-backed test isolation across xdist runs (#543)

## Summary

This PR reduces flaky Redis-backed service tests by fixing test
isolation issues in the test suite rather than changing product code.

The failures were caused by Redis state leaking across tests during
`pytest -n auto` runs. Several tests reused fixed Redis resource names,
and some shared HNSW index fixtures created Redis indexes without
deleting them. That made unrelated tests on the same xdist worker
sensitive to execution order and leftover state, which explains why some
matrix failures passed on retry.

## Problem

The main issue was cross-test persistence in Redis:

- Shared fixtures used Redis index names and prefixes scoped only to
`worker_id`.
- `hnsw_index` and `async_hnsw_index` created indexes but did not tear
them down.
- Several integration tests created histories, routers, caches, or
indexes with fixed names like `test_app`, `float64 history`,
`test_pass_through_dtype`, and `test_multi_prefix`.

In CI, that meant one test could leave Redis resources behind for
another test running later on the same worker. The result was
intermittent failures that were noisy, hard to trust, and often
disappeared on rerun.

## Solution

This PR makes Redis-backed test resources unique per test and restores
missing cleanup.

A shared pytest helper now generates per-test Redis names using the
worker id plus the current test node. That helper is used in the shared
search-index fixtures and in the highest-risk integration tests that
previously reused fixed names. The HNSW fixtures now delete their
indexes after yielding, and tests that instantiate Redis-backed objects
directly now clean them up explicitly.

The semantic router tests also no longer need the existing “flaky test”
skips because their Redis setup is now deterministic under unique names.

## Changes

- Added a per-test Redis naming fixture in `tests/conftest.py`
- Updated shared flat/HNSW index fixtures to use per-test index names
and prefixes
- Added teardown to `hnsw_index` and `async_hnsw_index`
- Replaced fixed Redis names in `test_message_history.py`
- Replaced fixed Redis names in `test_semantic_router.py`
- Added explicit cleanup for direct `SemanticMessageHistory` and
`SemanticRouter` instances
- Updated embed cache fixtures to use per-test cache names
- Updated the multiple-prefix existing-index test to use a unique index
name

## Scope

- Test-only changes
- No production API changes
- No CI workflow changes
- No coverage reduction

## Verification

- Targeted Redis-backed suites passed under `pytest -n auto`
- Repeated reruns showed the isolation fixes holding at the assertion
level
- There is still a separate pytest master shutdown hang during some
repeated xdist reruns, but that is distinct from the Redis leakage fixed
here

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Low Risk**
> Low risk: changes are limited to test fixtures/assertions, mainly
improving Redis resource isolation and teardown to reduce CI flakiness
under `pytest-xdist`. Main risk is unintended name/cleanup changes
masking real failures or leaving less coverage for shared-state
scenarios.
> 
> **Overview**
> Improves Redis-backed test isolation under `pytest -n` by introducing
a `redis_test_name` fixture (worker id + test node hash) and switching
indices/caches/histories/routers to use per-test Redis names instead of
fixed or worker-scoped ones.
> 
> Adds missing teardown for HNSW index fixtures and wraps several
integration tests in explicit `clear()`/`delete()` cleanup blocks; also
removes prior *flaky test* skips and updates a hybrid-search assertion
to match the actual number of loaded docs.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
7e2afe8. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: vishal-bala

tests/conftest.py

@@ -1,5 +1,7 @@
+import hashlib
 import logging
 import os
+import re
 import subprocess
 import sys
 from datetime import datetime, timezone
@@ -33,6 +35,18 @@ def worker_id(request):
     return workerinput.get("workerid", "master")
 
 
+@pytest.fixture
+def redis_test_name(worker_id, request):
+    """Build a per-test Redis resource name stable within a test function."""
+    node_hash = hashlib.sha1(request.node.nodeid.encode("utf-8")).hexdigest()[:10]
+
+    def make_name(base: str) -> str:
+        slug = re.sub(r"[^0-9A-Za-z]+", "_", base).strip("_").lower()
+        return f"{slug or 'redis_resource'}_{worker_id}_{node_hash}"
+
+    return make_name
+
+
 @pytest.fixture(autouse=True)
 def set_tokenizers_parallelism():
     """Disable tokenizers parallelism in tests to avoid deadlocks"""
@@ -466,17 +480,19 @@ def pytest_collection_modifyitems(
 
 
 @pytest.fixture
-def flat_index(sample_data, redis_url, worker_id):
+def flat_index(sample_data, redis_url, redis_test_name):
     """
     A fixture that uses the "flag" algorithm for its vector field.
     """
 
     # construct a search index from the schema
+    index_name = redis_test_name("user_index")
+    index_prefix = redis_test_name("v1")
     index = SearchIndex.from_dict(
         {
             "index": {
-                "name": f"user_index_{worker_id}",
-                "prefix": f"v1_{worker_id}",
+                "name": index_name,
+                "prefix": index_prefix,
                 "storage_type": "hash",
             },
             "fields": [
@@ -521,17 +537,19 @@ def hash_preprocess(item: dict) -> dict:
 
 
 @pytest.fixture
-async def async_flat_index(sample_data, redis_url, worker_id):
+async def async_flat_index(sample_data, redis_url, redis_test_name):
     """
     A fixture that uses the "flag" algorithm for its vector field.
     """
 
     # construct a search index from the schema
+    index_name = redis_test_name("user_index")
+    index_prefix = redis_test_name("v1")
     index = AsyncSearchIndex.from_dict(
         {
             "index": {
-                "name": f"user_index_{worker_id}",
-                "prefix": f"v1_{worker_id}",
+                "name": index_name,
+                "prefix": index_prefix,
                 "storage_type": "hash",
             },
             "fields": [
@@ -576,16 +594,18 @@ def hash_preprocess(item: dict) -> dict:
 
 
 @pytest.fixture
-async def async_hnsw_index(sample_data, redis_url, worker_id):
+async def async_hnsw_index(sample_data, redis_url, redis_test_name):
     """
     A fixture that uses the "hnsw" algorithm for its vector field.
     """
 
+    index_name = redis_test_name("user_index")
+    index_prefix = redis_test_name("v1")
     index = AsyncSearchIndex.from_dict(
         {
             "index": {
-                "name": f"user_index_{worker_id}",
-                "prefix": f"v1_{worker_id}",
+                "name": index_name,
+                "prefix": index_prefix,
                 "storage_type": "hash",
             },
             "fields": [
@@ -625,18 +645,23 @@ def hash_preprocess(item: dict) -> dict:
     # run the test
     yield index
 
+    # clean up
+    await index.delete(drop=True)
+
 
 @pytest.fixture
-def hnsw_index(sample_data, redis_url, worker_id):
+def hnsw_index(sample_data, redis_url, redis_test_name):
     """
     A fixture that uses the "hnsw" algorithm for its vector field.
     """
 
+    index_name = redis_test_name("user_index")
+    index_prefix = redis_test_name("v1")
     index = SearchIndex.from_dict(
         {
             "index": {
-                "name": f"user_index_{worker_id}",
-                "prefix": f"v1_{worker_id}",
+                "name": index_name,
+                "prefix": index_prefix,
                 "storage_type": "hash",
             },
             "fields": [
@@ -676,6 +701,9 @@ def hash_preprocess(item: dict) -> dict:
     # run the test
     yield index
 
+    # clean up
+    index.delete(drop=True)
+
 
 # Version checking utilities
 def get_redis_version(client):

tests/integration/test_embedcache.py

@@ -11,10 +11,10 @@
 
 
 @pytest.fixture
-def cache(redis_url, worker_id):
+def cache(redis_url, redis_test_name):
     """Basic EmbeddingsCache fixture with cleanup."""
     cache_instance = EmbeddingsCache(
-        name=f"test_embed_cache_{worker_id}",
+        name=redis_test_name("test_embed_cache"),
         redis_url=redis_url,
     )
     yield cache_instance
@@ -23,10 +23,10 @@ def cache(redis_url, worker_id):
 
 
 @pytest.fixture
-def cache_with_ttl(redis_url):
+def cache_with_ttl(redis_url, redis_test_name):
     """EmbeddingsCache with TTL setting."""
     cache_instance = EmbeddingsCache(
-        name="test_ttl_cache",
+        name=redis_test_name("test_ttl_cache"),
         ttl=2,  # 2 second TTL for testing expiration
         redis_url=redis_url,
     )
@@ -36,10 +36,10 @@ def cache_with_ttl(redis_url):
 
 
 @pytest.fixture
-def cache_with_redis_client(client):
+def cache_with_redis_client(client, redis_test_name):
     """EmbeddingsCache with provided Redis client."""
     cache_instance = EmbeddingsCache(
-        name="test_client_cache",
+        name=redis_test_name("test_client_cache"),
         redis_client=client,
     )
     yield cache_instance

tests/integration/test_hybrid.py

@@ -136,7 +136,7 @@ def test_hybrid_query(index):
 
     results = index.query(hybrid_query)
     assert isinstance(results, list)
-    assert len(results) == 10
+    assert len(results) == 7  # all docs in the index (see `multi_vector_data` fixture)
     for doc in results:
         assert doc["user"] in [
             "john",