from common import *
from includes import *

# Constant value for _COORD_DISPATCH_TIME argument in internal commands
COORD_DISPATCH_TIME = '1000000'  # 1ms in nanoseconds

def remove_warnings(result):
    """Remove any warnings from the result and return the rest"""

    if isinstance(result, list):
        warnings_index = result.index('warnings') if 'warnings' in result else -1
        if warnings_index != -1:
            return result[:warnings_index]
        return result
    if isinstance(result, dict):
        if 'warnings' in result:
            del result['warnings']
        return result
    return result

def setup_hybrid_test_data(env):
    """Setup test data based on the provided scenario"""
    # Create index with text and vector fields
    env.expect('FT.CREATE', 'idx', 'SCHEMA',
               'description', 'TEXT',
               'embedding', 'VECTOR', 'FLAT', '6', 'TYPE', 'FLOAT32', 'DIM', '2', 'DISTANCE_METRIC', 'L2').ok()

    # Add test documents with embeddings
    conn = getConnectionByEnv(env)
    conn.execute_command('HSET', 'doc:1', 'description', 'red shoes',
                        'embedding', create_np_array_typed([0.0, 0.0], 'FLOAT32').tobytes())
    conn.execute_command('HSET', 'doc:2', 'description', 'red running shoes',
                        'embedding', create_np_array_typed([1.0, 0.0], 'FLOAT32').tobytes())
    conn.execute_command('HSET', 'doc:3', 'description', 'running gear',
                        'embedding', create_np_array_typed([0.0, 1.0], 'FLOAT32').tobytes())
    conn.execute_command('HSET', 'doc:4', 'description', 'blue shoes',
                        'embedding', create_np_array_typed([1.0, 1.0], 'FLOAT32').tobytes())

    # Mark as internal client for _FT.HYBRID command
    env.cmd('DEBUG', 'MARK-INTERNAL-CLIENT')


def read_cursor_completely_resp3(env, index_name, cursor_id, batch_callback=None):
    """Read all results from a cursor and return them (RESP 3 format)

    Args:
        env: Test environment
        index_name: Name of the index
        cursor_id: Cursor ID to read from
        batch_callback: Optional function called for each batch with (batch_results, cursor_response)

    Returns:
        list: All results from the cursor as dicts with '__key' and optionally 'score' fields
    """
    if cursor_id == 0:
        return []

    all_results = []
    current_cursor = cursor_id

    while current_cursor != 0:
        cursor_response = env.execute_command('_FT.CURSOR', 'READ', index_name, current_cursor)
        # RESP 3 format: [{'results': [...], ...}, cursor_id]
        results_dict = cursor_response[0]
        current_cursor = cursor_response[1]
        batch_results = results_dict['results']

        # Call batch callback if provided
        if batch_callback:
            batch_callback(batch_results, cursor_response)

        # Extract document keys and scores from cursor results
        for result in batch_results:
            score = result.get('score')
            key = result.get('extra_attributes', {}).get('__key', '')
            all_results.append({
                        'key': key,
                        'score': score
                    })

    return sorted(all_results, key=lambda x: x['key'] if isinstance(x, dict) else x)


def read_cursor_completely_resp2(env, index_name, cursor_id, batch_callback=None):
    """Read all results from a cursor and return them (RESP 2 format)

    Args:
        env: Test environment
        index_name: Name of the index
        cursor_id: Cursor ID to read from
        batch_callback: Optional function called for each batch with (batch_results, cursor_response)

    Returns:
        list: All results from the cursor as document key strings (RESP 2 doesn't include scores)
    """
    if cursor_id == 0:
        return []

    all_results = []
    current_cursor = cursor_id

    while current_cursor != 0:
        cursor_response = env.execute_command('_FT.CURSOR', 'READ', index_name, current_cursor)

        # RESP 2 format: [[count, result1, result2, ...], next_cursor_id]
        results_array = cursor_response[0]
        current_cursor = cursor_response[1]
        batch_results = results_array[1:]  # Skip the count at index 0

        # Call batch callback if provided
        if batch_callback:
            batch_callback(batch_results, cursor_response)

        # Extract document keys from cursor results (RESP 2 doesn't include scores)
        for result in batch_results:
            result_dict = dict(zip(result[::2], result[1::2]))
            key = result_dict.get('__key')
            if key is not None:
                all_results.append(key)

    return sorted(all_results)


def read_cursor_completely(env, index_name, cursor_id, batch_callback=None, protocol=None):
    """Read all results from a cursor and return them (auto-detect RESP format)

    Args:
        env: Test environment
        index_name: Name of the index
        cursor_id: Cursor ID to read from
        batch_callback: Optional function called for each batch with (batch_results, cursor_response)

    Returns:
        list: All results from the cursor as dicts with '__key' and optionally 'score' fields
    """
    # Use RESP 3 by default since that's what most tests use
    if protocol is not None and protocol == 2:
        return read_cursor_completely_resp2(env, index_name, cursor_id, batch_callback)
    else:
        return read_cursor_completely_resp3(env, index_name, cursor_id, batch_callback)


def get_shard_slot_ranges(env):
    """Get slot ranges for each shard in cluster mode, or full range for standalone"""
    if not env.isCluster():
        # Standalone mode: single shard owns all slots
        return [(0, generate_slots())]

    # Cluster mode: get actual slot ranges from cluster topology
    cluster_info = env.cmd('CLUSTER', 'SLOTS')
    shard_ranges = []

    for shard_id, slot_info in enumerate(cluster_info):
        # Each slot_info is a list like:
        # [start_slot, end_slot, [ip, port, node_id, []], ...]

        start_slot = slot_info[0]
        end_slot = slot_info[1]

        # Generate the slots data for this range
        slots_data = generate_slots(range(start_slot, end_slot + 1))
        shard_ranges.append((shard_id + 1, slots_data))

    return shard_ranges


def test_basic_hybrid_internal_withcursor(env):
    """Test basic _FT.HYBRID command with WITHCURSOR functionality

    Expected behavior when fixed:
    - Should return a map with VSIM and SEARCH cursor IDs
    - Format: ['VSIM', cursor_id, 'SEARCH', cursor_id]
    - Both cursor IDs should be valid integers/strings
    """
    setup_hybrid_test_data(env)

    # Get slot ranges for each shard
    shard_ranges = get_shard_slot_ranges(env)

    # Test each shard with its appropriate slot range
    for shard_id, slots_data in shard_ranges:
        # Execute _FT.HYBRID command with WITHCURSOR using shard-specific slots
        query_vec = create_np_array_typed([0.0, 0.0], 'FLOAT32')

        if env.isCluster():
            # In cluster mode, send to specific shard
            shard_conn = env.getConnection(shardId=shard_id)
            shard_conn.execute_command('DEBUG', 'MARK-INTERNAL-CLIENT')
            result = shard_conn.execute_command('_FT.HYBRID', 'idx', 'SEARCH', '@description:running',
                                              'VSIM', '@embedding', '$BLOB',
                                              'WITHCURSOR', '_SLOTS_INFO', slots_data, 'PARAMS', '2', 'BLOB', query_vec.tobytes(),
                                              '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME)
        else:
            # In standalone mode, send to main connection
            result = env.cmd('_FT.HYBRID', 'idx', 'SEARCH', '@description:running',
                           'VSIM', '@embedding', '$BLOB',
                           'WITHCURSOR', '_SLOTS_INFO', slots_data, 'PARAMS', '2', 'BLOB', query_vec.tobytes(),
                           '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME)

        # Should return a map with VSIM and SEARCH cursor IDs
        env.assertTrue(isinstance(result, list))
        env.assertTrue(len(result) > 0)

        # Convert list to dict for easier access
        result_dict = to_dict(remove_warnings(result))

        # Should have VSIM and SEARCH cursor IDs
        env.assertIn('VSIM', result_dict)
        env.assertIn('SEARCH', result_dict)

        # Both cursor IDs should be valid integers
        vsim_cursor = result_dict['VSIM']
        search_cursor = result_dict['SEARCH']
        env.assertTrue(isinstance(vsim_cursor, (int, str)))
        env.assertTrue(isinstance(search_cursor, (int, str)))


@skip(cluster=True)
def test_hybrid_internal_without_slots_info(env):
    """_FT.HYBRID without _SLOTS_INFO (as sent by coordinators older than 8.4) must not
    fail; the shard falls back to its local slots (MOD-16047)."""
    setup_hybrid_test_data(env)

    query_vec = create_np_array_typed([0.0, 0.0], 'FLOAT32')
    query = ('_FT.HYBRID', 'idx', 'SEARCH', '@description:running',
             'VSIM', '@embedding', '$BLOB',
             'WITHCURSOR', 'PARAMS', '2', 'BLOB', query_vec.tobytes(),
             '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME)

    # Should return a map with VSIM and SEARCH cursor IDs, like a fully-specified query
    result = env.cmd(*query)
    result_dict = to_dict(remove_warnings(result))
    env.assertIn('VSIM', result_dict)
    env.assertIn('SEARCH', result_dict)


@skip(cluster=True)
def test_hybrid_internal_without_coord_dispatch_time(env):
    """_FT.HYBRID without _COORD_DISPATCH_TIME (as sent by coordinators older than 8.6)
    must not fail."""
    setup_hybrid_test_data(env)

    query_vec = create_np_array_typed([0.0, 0.0], 'FLOAT32')
    query = ('_FT.HYBRID', 'idx', 'SEARCH', '@description:running',
             'VSIM', '@embedding', '$BLOB',
             'WITHCURSOR', '_SLOTS_INFO', generate_slots(),
             'PARAMS', '2', 'BLOB', query_vec.tobytes())

    result = env.cmd(*query)
    result_dict = to_dict(remove_warnings(result))
    env.assertIn('VSIM', result_dict)
    env.assertIn('SEARCH', result_dict)


def test_hybrid_internal_with_count_parameter(env):
    """Test _FT.HYBRID with WITHCURSOR and COUNT parameter"""
    setup_hybrid_test_data(env)

    slot_ranges = get_shard_slot_ranges(env)

    # Execute with COUNT parameter set to 2 using direct vector specification
    count_param = 2
    query_vec = create_np_array_typed([0.0, 0.0], 'FLOAT32')
    for shard_id, slots_data in slot_ranges:
        if env.isCluster():
            # In cluster mode, send to specific shard
            shard_conn = env.getConnection(shardId=shard_id)
            shard_conn.execute_command('DEBUG', 'MARK-INTERNAL-CLIENT')
            result = shard_conn.execute_command('_FT.HYBRID', 'idx', 'SEARCH', '@description:running',
                                              'VSIM', '@embedding', '$BLOB',
                                              'WITHCURSOR', 'COUNT', str(count_param), '_SLOTS_INFO', slots_data, 'PARAMS', '2', 'BLOB', query_vec.tobytes(),
                                              '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME)
        else:
            result = env.cmd('_FT.HYBRID', 'idx', 'SEARCH', '@description:running',
                              'VSIM', '@embedding', '$BLOB', 'WITHCURSOR', 'COUNT', str(count_param), '_SLOTS_INFO', generate_slots(), 'PARAMS', '2', 'BLOB', query_vec.tobytes(),
                              '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME)

        # Should return a map with cursor IDs
        env.assertTrue(isinstance(result, list))
        result = remove_warnings(result)
        result_dict = dict(zip(result[::2], result[1::2]))

        # Should have both cursor types
        env.assertIn('VSIM', result_dict)
        env.assertIn('SEARCH', result_dict)

        # Test reading from cursors with COUNT parameter using callback
        def validate_batch_size(batch_results, _cursor_response):
            """Callback to validate that each batch respects the COUNT parameter"""
            # The key test: number of results in each batch should be <= COUNT parameter
            env.assertTrue(len(batch_results) <= count_param)
        for cursor_id in result_dict.values():
            if cursor_id != 0:  # Only test active cursors
                # Use common function with callback to validate COUNT behavior
                if env.isCluster():
                    results = read_cursor_completely(shard_conn, 'idx', cursor_id, validate_batch_size, protocol=getattr(env, 'protocol', None))
                else:
                    results = read_cursor_completely(env, 'idx', cursor_id, validate_batch_size, protocol=getattr(env, 'protocol', None))
                env.assertTrue(isinstance(results, list))


def test_hybrid_internal_cursor_interaction(env):
    """Test reading from both VSIM and SEARCH cursors and compare with equivalent FT.SEARCH commands"""
    setup_hybrid_test_data(env)

    # Get slot ranges for each shard
    shard_ranges = get_shard_slot_ranges(env)
    # Execute the hybrid command with cursors using direct vector specification
    query_vec = create_np_array_typed([1.0, 0.0], 'FLOAT32')

    cursor_results_accum_text = []
    cursor_results_accum_vector = []

    # Test each shard with its appropriate slot range
    for shard_id, slots_data in shard_ranges:
        # Execute the hybrid command with cursors using shard-specific slots
        query_vec = create_np_array_typed([1.0, 0.0], 'FLOAT32')

        if env.isCluster():
            # In cluster mode, send to specific shard
            shard_conn = env.getConnection(shardId=shard_id)
            shard_conn.execute_command('DEBUG', 'MARK-INTERNAL-CLIENT')
            hybrid_result = shard_conn.execute_command('_FT.HYBRID', 'idx', 'SEARCH', '@description:shoes',
                                                     'VSIM', '@embedding', '$BLOB',
                                                     'WITHCURSOR', '_SLOTS_INFO', slots_data, 'PARAMS', '2', 'BLOB', query_vec.tobytes(),
                                                     '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME)
        else:
            # In standalone mode, send to main connection
            hybrid_result = env.cmd('_FT.HYBRID', 'idx', 'SEARCH', '@description:shoes',
                                  'VSIM', '@embedding', '$BLOB',
                                  'WITHCURSOR', '_SLOTS_INFO', slots_data, 'PARAMS', '2', 'BLOB', query_vec.tobytes(),
                                  '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME)
        hybrid_result = remove_warnings(hybrid_result)
        # Should return a map with cursor IDs
        env.assertTrue(isinstance(hybrid_result, list))
        result_dict = dict(zip(hybrid_result[::2], hybrid_result[1::2]))

        # Should have both cursor types
        env.assertIn('VSIM', result_dict)
        env.assertIn('SEARCH', result_dict)

        # Read from cursors and collect results using common function
        cursor_results = {}
        for cursor_type, cursor_id in result_dict.items():
            if env.isCluster():
                cursor_results[cursor_type] = read_cursor_completely(shard_conn, 'idx', cursor_id, protocol=getattr(env, 'protocol', None))
            else:
                cursor_results[cursor_type] = read_cursor_completely(env, 'idx', cursor_id, protocol=getattr(env, 'protocol', None))
        if 'SEARCH' in cursor_results:
            cursor_results_accum_text.append(cursor_results['SEARCH'])
        if 'VSIM' in cursor_results:
            cursor_results_accum_vector.append(cursor_results['VSIM'])

    text_search_result = env.cmd('FT.SEARCH', 'idx', '@description:shoes', 'DIALECT', '2', 'RETURN', '0')
    vector_search_result = env.cmd('FT.SEARCH', 'idx', '*=>[KNN 10 @embedding $vec_param]', 'DIALECT', '2',
                                  'PARAMS', '2', 'vec_param', query_vec.tobytes(), 'RETURN', '0')

    # Extract document keys from expected results (RETURN 0 format)
    def extract_doc_keys(search_result):
        """Extract document keys from FT.SEARCH result format with RETURN 0"""
        if len(search_result) < 2:
            return []

        doc_keys = []
        # Skip the count (first element), remaining elements are just document keys
        for i in range(1, len(search_result)):
            doc_keys.append(search_result[i])
        return sorted(doc_keys)

    expected_text_docs = extract_doc_keys(text_search_result)
    expected_vector_docs = extract_doc_keys(vector_search_result)
    sorted_cursor_results_accum_text = sorted([doc for sublist in cursor_results_accum_text for doc in sublist])
    sorted_cursor_results_accum_vector = sorted([doc for sublist in cursor_results_accum_vector for doc in sublist])
    env.assertEqual(sorted_cursor_results_accum_text, expected_text_docs)
    env.assertEqual(sorted_cursor_results_accum_vector, expected_vector_docs)


def test_hybrid_internal_cursor_with_scores():
    """Test reading from both VSIM and SEARCH cursors with WITHSCORES and compare with equivalent FT.SEARCH commands"""
    env = Env(protocol=3, moduleArgs='DEFAULT_DIALECT 2')
    setup_hybrid_test_data(env)
    slots = generate_slots(range(0, int((2**14)/env.shardsCount)))

    # Execute the hybrid command with cursors
    query_vec = create_np_array_typed([1.0, 0.0], 'FLOAT32')
    hybrid_cursor_dict = env.cmd('_FT.HYBRID', 'idx', 'SEARCH', '@description:shoes',
                           'VSIM', '@embedding', '$vec_param', 'KNN', '2', 'K', '10',
                           'WITHCURSOR', 'WITHSCORES',
                           'PARAMS', '2', 'vec_param', query_vec.tobytes(), '_SLOTS_INFO', slots,
                           '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME)

    hybrid_cursor_dict = remove_warnings(hybrid_cursor_dict)
    # Should return a map with cursor IDs
    env.assertTrue(isinstance(hybrid_cursor_dict, dict))

    # Should have both cursor types
    env.assertIn('VSIM', hybrid_cursor_dict)
    env.assertIn('SEARCH', hybrid_cursor_dict)

    # Read from cursors and collect results using common function
    cursor_results = {}
    for cursor_type, cursor_id in hybrid_cursor_dict.items():
        cursor_results[cursor_type] = read_cursor_completely(env, 'idx', cursor_id, protocol=getattr(env, 'protocol', None))

    for cursor_type, cursor_result in cursor_results.items():
        for result in cursor_result:
            env.assertIn('key', result)
            env.assertIn('doc', result['key'])
            env.assertIn('score', result)
            env.assertTrue(isinstance(result['score'], (int, float)))


def test_hybrid_internal_with_params(env):
    """Test _FT.HYBRID with WITHCURSOR and PARAMS functionality"""
    setup_hybrid_test_data(env)

    # Get slot ranges for each shard
    shard_ranges = get_shard_slot_ranges(env)

    cursor_results_accum_text = []
    cursor_results_accum_vector = []

    # Test each shard with its appropriate slot range
    for shard_id, slots_data in shard_ranges:
        # Test with PARAMS for both text and vector parts
        query_vec = create_np_array_typed([1.0, 0.0], 'FLOAT32')

        # Execute hybrid command with shard-specific slots
        if env.isCluster():
            # In cluster mode, send to specific shard
            shard_conn = env.getConnection(shardId=shard_id)
            shard_conn.execute_command('DEBUG', 'MARK-INTERNAL-CLIENT')
            hybrid_result = shard_conn.execute_command('_FT.HYBRID', 'idx', 'SEARCH', '@description:($term)',
                                                     'VSIM', '@embedding', '$BLOB', 'WITHCURSOR',
                                                     'PARAMS', '4', 'term', 'shoes', 'BLOB', query_vec.tobytes(),'_SLOTS_INFO', slots_data,
                                                     '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME)
        else:
            # In standalone mode, send to main connection
            hybrid_result = env.cmd('_FT.HYBRID', 'idx', 'SEARCH', '@description:($term)',
                                  'VSIM', '@embedding', '$BLOB', 'WITHCURSOR',
                                  'PARAMS', '4', 'term', 'shoes', 'BLOB', query_vec.tobytes(), '_SLOTS_INFO', slots_data,
                                  '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME)
        hybrid_result = remove_warnings(hybrid_result)

        # Should return cursor map
        env.assertTrue(isinstance(hybrid_result, list))
        result_dict = dict(zip(hybrid_result[::2], hybrid_result[1::2]))
        env.assertIn('VSIM', result_dict)
        env.assertIn('SEARCH', result_dict)

        # Read cursor results and compare with expected results
        cursor_results = {}
        for cursor_type, cursor_id in result_dict.items():
            if env.isCluster():
                cursor_results[cursor_type] = read_cursor_completely(shard_conn, 'idx', cursor_id, protocol=getattr(env, 'protocol', None))
            else:
                cursor_results[cursor_type] = read_cursor_completely(env, 'idx', cursor_id, protocol=getattr(env, 'protocol', None))

        # Verify that parameterized queries work correctly
        cursor_results_accum_text.append(cursor_results['SEARCH'])
        cursor_results_accum_vector.append(cursor_results['VSIM'])

    # Get expected results from equivalent parameterized FT.SEARCH commands
    text_search_result = env.cmd('FT.SEARCH', 'idx', '@description:($term)', 'DIALECT', '2',
                                'PARAMS', '2', 'term', 'shoes', 'RETURN', '0')
    vector_search_result = env.cmd('FT.SEARCH', 'idx', '*=>[KNN 10 @embedding $vec_param]', 'DIALECT', '2',
                                  'PARAMS', '2', 'vec_param', query_vec.tobytes(), 'RETURN', '0')

    # Extract expected document keys
    def extract_doc_keys(search_result):
        return sorted(search_result[1:]) if len(search_result) > 1 else []

    expected_text_docs = extract_doc_keys(text_search_result)
    expected_vector_docs = extract_doc_keys(vector_search_result)
    sorted_cursor_results_accum_text = sorted([doc for sublist in cursor_results_accum_text for doc in sublist])
    sorted_cursor_results_accum_vector = sorted([doc for sublist in cursor_results_accum_vector for doc in sublist])
    env.assertEqual(sorted_cursor_results_accum_text, expected_text_docs)
    env.assertEqual(sorted_cursor_results_accum_vector, expected_vector_docs)


def test_hybrid_internal_error_cases(env):
    """Test error cases with _FT.HYBRID (without WITHCURSOR)"""
    setup_hybrid_test_data(env)

    # Test with non-existent index using direct vector specification
    query_vec = create_np_array_typed([0.0, 0.0], 'FLOAT32')
    env.expect('_FT.HYBRID', 'nonexistent', 'SEARCH', '@description:running',
               'VSIM', '@embedding', '$BLOB', '_SLOTS_INFO', generate_slots(range(0, 0)), 'PARAMS', '2', 'BLOB', query_vec.tobytes(),
               '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME).error().contains('SEARCH_INDEX_NOT_FOUND Index not found: nonexistent')

    # Test with invalid vector field using direct vector specification
    env.expect('_FT.HYBRID', 'idx', 'SEARCH', '@description:running',
               'VSIM', '@nonexistent', '$BLOB', '_SLOTS_INFO', generate_slots(range(0, 0)), 'PARAMS', '2', 'BLOB', query_vec.tobytes(),
               '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME).error().contains('Unknown field `nonexistent`')

    # Test with bad slots data
    env.expect('_FT.HYBRID', 'idx', 'SEARCH', '@description:running',
               'VSIM', '@embedding', '$BLOB', '_SLOTS_INFO', 'BAD_SLOTS_DATA', 'PARAMS', '2', 'BLOB', query_vec.tobytes(),
               '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME).error().contains('Failed to deserialize _SLOTS_INFO data')
    # Edge case: Test syntax error after parsing _SLOTS_INFO (for coverage and memory leaks)
    env.expect('_FT.HYBRID', 'idx', 'SEARCH', '@description:running',
               'VSIM', '@embedding', '$BLOB', '_SLOTS_INFO', generate_slots(range(0, 0)), 'PARAMS', '2', 'BLOB', query_vec.tobytes(), 'INVALID_SYNTAX',
               '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME).error().contains('Unknown argument')


def test_hybrid_internal_cursor_limit(env):
    """Test _FT.HYBRID cursor limit per shard

    A single _FT.HYBRID command tries to create two cursors (VSIM and SEARCH).
    When INDEX_CURSOR_LIMIT is set to 1, this should fail with 'Failed to allocate enough cursors' error.
    """
    # Set cursor limit to 1 for this test
    env.cmd('CONFIG', 'SET', 'search-index-cursor-limit', '1')

    setup_hybrid_test_data(env)

    # _FT.HYBRID command should fail because it tries to create 2 cursors but limit is 1
    query_vec = create_np_array_typed([0.0, 0.0], 'FLOAT32')
    env.expect('_FT.HYBRID', 'idx', 'SEARCH', '@description:running',
               'VSIM', '@embedding', '$BLOB',
               'PARAMS', '2', 'BLOB', query_vec.tobytes(),
               'WITHCURSOR', '_SLOTS_INFO', generate_slots(range(0, 0)),
               '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME).error().contains('INDEX_CURSOR_LIMIT of 1 has been reached for an index')


def test_hybrid_internal_empty_search_results(env):
    """Test _FT.HYBRID when search subquery returns no results

    This test verifies behavior when the text search part finds no matching documents,
    while the vector similarity part can still return results.
    """
    setup_hybrid_test_data(env)

    # Search for a term that doesn't exist in any document
    query_vec = create_np_array_typed([0.0, 0.0], 'FLOAT32')

    for shard_id, slots_data in get_shard_slot_ranges(env):
        if env.isCluster():
            # In cluster mode, send to specific shard
            shard_conn = env.getConnection(shardId=shard_id)
            shard_conn.execute_command('DEBUG', 'MARK-INTERNAL-CLIENT')
            hybrid_result = shard_conn.execute_command('_FT.HYBRID', 'idx', 'SEARCH', '@description:nonexistent',
                                                     'VSIM', '@embedding', '$BLOB',  'WITHCURSOR',
                                                     '_SLOTS_INFO', slots_data, 'PARAMS', '2', 'BLOB', query_vec.tobytes(),
                                                     '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME)
        else:
            # In standalone mode, send to main connection
            hybrid_result = env.cmd('_FT.HYBRID', 'idx', 'SEARCH', '@description:nonexistent',
                                  'VSIM', '@embedding', '$BLOB',
                                  'WITHCURSOR', '_SLOTS_INFO', slots_data, 'PARAMS', '2', 'BLOB', query_vec.tobytes(),
                                  '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME)

        hybrid_result = remove_warnings(hybrid_result)
        # Should return a map with cursor IDs
        env.assertTrue(isinstance(hybrid_result, list))
        result_dict = dict(zip(hybrid_result[::2], hybrid_result[1::2]))

        # Should have both cursor types
        env.assertIn('VSIM', result_dict)
        env.assertIn('SEARCH', result_dict)

        # Verify that text search returns no results
        text_search_result = env.cmd('FT.SEARCH', 'idx', '@description:nonexistent', 'DIALECT', '2', 'RETURN', '0')
        env.assertEqual(text_search_result[0], 0)  # Should have 0 results

        # Verify that vector search still returns results
        vector_search_result = env.cmd('FT.SEARCH', 'idx', '*=>[KNN 10 @embedding $vec_param]', 'DIALECT', '2',
                                      'PARAMS', '2', 'vec_param', query_vec.tobytes(), 'RETURN', '0')
        env.assertTrue(vector_search_result[0] > 0)  # Should have results

        # Read from cursors and verify behavior
        cursor_results = {}
        for cursor_type, cursor_id in result_dict.items():
            if env.isCluster():
                cursor_results[cursor_type] = read_cursor_completely(shard_conn, 'idx', cursor_id, protocol=getattr(env, 'protocol', None))
            else:
                cursor_results[cursor_type] = read_cursor_completely(env, 'idx', cursor_id, protocol=getattr(env, 'protocol', None))

        # SEARCH cursor should return empty results
        env.assertEqual(cursor_results['SEARCH'], [])

        # VSIM cursor should return some results
        env.assertTrue(len(cursor_results['VSIM']) > 0)

@skip(cluster=True)
def test_hybrid_internal_withcursor_with_load():
    """Test basic _FT.HYBRID command with WITHCURSOR functionality and explicit load of __key and description
    """
    env = Env(enableDebugCommand=True)
    setup_hybrid_test_data(env)

    # Execute _FT.HYBRID command with WITHCURSOR using direct vector specification
    query_vec = create_np_array_typed([0.0, 0.0], 'FLOAT32')
    result = env.cmd('_FT.HYBRID', 'idx', 'SEARCH', '@description:running',
                     'VSIM', '@embedding', '$BLOB',
                     'PARAMS', '2', 'BLOB', query_vec.tobytes(),
                     'LOAD', '2', '@__key', '@description',
                     'WITHCURSOR', '_SLOTS_INFO', generate_slots(),
                     '_COORD_DISPATCH_TIME', COORD_DISPATCH_TIME)

    # Should return a map with VSIM and SEARCH cursor IDs
    env.assertTrue(isinstance(result, list))
    env.assertTrue(len(result) > 0)

    # Convert list to dict for easier access
    result_dict = dict(zip(result[::2], result[1::2]))

    # Should have VSIM and SEARCH cursor IDs
    env.assertIn('VSIM', result_dict)
    env.assertIn('SEARCH', result_dict)

    # Both cursor IDs should be valid integers
    vsim_cursor = result_dict['VSIM']
    search_cursor = result_dict['SEARCH']
    env.assertTrue(isinstance(vsim_cursor, (int, str)))
    env.assertTrue(isinstance(search_cursor, (int, str)))

    search_cursor_results = read_cursor_completely(env, 'idx', search_cursor, protocol=env.protocol)
    env.assertEqual(search_cursor_results, ['doc:2', 'doc:3'])

    vsim_cursor_results = read_cursor_completely(env, 'idx', vsim_cursor, protocol=getattr(env, 'protocol', None))
    env.assertEqual(vsim_cursor_results, ['doc:1', 'doc:2', 'doc:3', 'doc:4'])