nlproxy/core/compressor.py

"""
Semantic sentence compression module using clustering-based redundancy removal.

This module implements adaptive semantic compression by grouping semantically
similar sentences and selecting representative centroids, reducing token count
while preserving core meaning and constraints.

Mathematical Foundations
------------------------
1. Embedding Variance Filtering:
    Given embeddings E ∈ ℝⁿˣᵈ, variance per sentence:
        varᵢ = (1/d) Σⱼ (eᵢⱼ - μᵢ)²  where μᵢ = mean(eᵢ)
    Sentences with varᵢ < τ_variance are discarded as low-information.
    Reference: Jolliffe, "Principal Component Analysis" [1]

2. Euclidean Distance Matrix:
    Dᵢⱼ = ||eᵢ - eⱼ||₂ = √(Σₖ (eᵢₖ - eⱼₖ)²)
    Computed via scipy.spatial.distance.cdist with O(n²·d) complexity.
    Reference: scipy.spatial.distance documentation [2]

3. Hierarchical Clustering (Ward's Method):
    Minimizes within-cluster variance at each merge:
        Δ(A,B) = [n_A·n_B / (n_A + n_B)] · ||μ_A - μ_B||₂²
    Time complexity: O(n³) naive, O(n²·d) with optimized linkage.
    Reference: Ward, "Hierarchical Grouping to Optimize an Objective Function" [3]

4. K-Means Clustering (MiniBatch variant):
    Objective: minimize Σᵢ ||eᵢ - c_{label(i)}||₂²
    MiniBatch: uses random subsets of size b << n per iteration.
    Time complexity: O(n·k·d·i) where i=iterations, k=clusters.
    Reference: Sculley, "Web-Scale K-Means Clustering" [4]

5. Adaptive Threshold via Percentile:
    cutoff = P(aggressiveness·100) of upper-triangular distance matrix
    Lower percentile → stricter merging → higher compression.
    Mathematical: cutoff = quantile({Dᵢⱼ : i < j}, q=aggressiveness)

6. Centroid Selection within Clusters:
    For cluster C with embeddings {e₁, ..., eₘ}:
        centroid μ = (1/m) Σᵢ eᵢ
        representative = argmin_{e∈C} ||e - μ||₂
    Ensures selected sentence is most central semantically.

Compression Ratio Formula
-------------------------
    R = 1 - (|S_compressed| / |S_original|) ∈ [0, 1]
    Where |S| = number of sentences after processing.

    Effective aggressiveness α_eff ∈ [0, 1] controls target ratio:
        n_clusters ≈ |S_normal| · (1 - α_eff)  (for K-Means)
        cutoff_distance ≈ P(α_eff·100)  (for Ward)

References
----------
[1] Jolliffe, I. T. (2002). Principal Component Analysis, 2nd ed. Springer.

[2] Virtanen, P., et al. (2020). SciPy 1.0: Fundamental Algorithms for 
    Scientific Computing in Python. Nature Methods.
    https://github.com/scipy/scipy

[3] Ward, J. H. (1963). Hierarchical Grouping to Optimize an Objective 
    Function. Journal of the American Statistical Association, 58(301), 236-244.
    https://doi.org/10.1080/01621459.1963.10500845

[4] Sculley, D. (2010). Web-Scale K-Means Clustering. WWW 2010.
    https://github.com/google-research/google-research/tree/master/sculley_kmeans

[5] Reimers, N., & Gurevych, I. (2019). Sentence-BERT: Sentence embeddings 
    using Siamese BERT-networks. EMNLP-IJCNLP 2019.
    https://github.com/UKPLab/sentence-transformers

Performance Characteristics
---------------------------
- _remove_low_variance(): O(n·d) where n=sentences, d=embedding_dim
- _compute_condensed_distance(): O(n²·d) for pairwise Euclidean distances
- _cluster_ward(): O(n²·d + n² log n) for linkage + tree cutting
- _cluster_kmeans(): O(n·k·d·i) with MiniBatch optimization (i≈10-100)
- compress() full pipeline: 
    * Small n (<200): O(n²·d) dominated by Ward clustering
    * Large n (≥200): O(n·k·d·i) dominated by K-Means
- Memory: O(n²) for distance matrix (Ward), O(n·d) for K-Means

Author: IntelliDeep Labs Team
License: BSL 1.1
"""

from __future__ import annotations

import asyncio
import logging
import time
from typing import Dict, List, Optional, Tuple

import numpy as np
from scipy.cluster.hierarchy import linkage, fcluster
from scipy.spatial.distance import cdist, squareform
from sklearn.cluster import MiniBatchKMeans


logger = logging.getLogger(__name__)



class SemanticCompressor:
    """
    Adaptive semantic compressor using clustering-based redundancy removal.

    This class reduces sentence count by grouping semantically similar sentences
    and selecting representative centroids, while preserving:
    - Protected placeholders (__PROT_*) from code/PII shielding
    - Domain-specific compression aggressiveness presets
    - Original sentence ordering for coherent reconstruction

    Key Features
    ------------
    - Dual clustering backends: Ward (hierarchical, precise) or K-Means (fast)
    - Auto-selection based on sentence count threshold
    - Variance-based filtering to discard low-information sentences
    - Adaptive distance cutoff via percentile of empirical distance distribution
    - Async support for non-blocking batch processing

    Compression Strategy
    --------------------
    1. Separate protected sentences (placeholders) from compressible content
    2. Filter out low-variance embeddings (noise reduction)
    3. Cluster remaining sentences using selected algorithm:
       * Ward: Hierarchical agglomerative clustering with variance minimization
       * K-Means: Partition-based clustering with MiniBatch optimization
    4. Select centroid sentence from each cluster (closest to mean embedding)
    5. Merge protected + compressed sentences preserving original order

    Mathematical Parameters
    -----------------------
    - aggressiveness ∈ [0, 1]: Target compression intensity
        * 0.0: No compression (keep all sentences)
        * 1.0: Maximum compression (one sentence per semantic group)
        * Default presets by mode: legal=0.25, finance=0.30, code=0.45, general=0.40
    
    - min_variance ≥ 0: Threshold for discarding low-information sentences
        * Computed as variance across embedding dimensions
        * Sentences with var < min_variance are considered noise
    
    - auto_method_threshold: Sentence count at which to switch from Ward to K-Means
        * Ward: O(n²) memory, precise for small n
        * K-Means: O(n) memory, scalable for large n

    Usage Example
    -------------
    >>> compressor = SemanticCompressor(mode="code", aggressiveness=0.3)
    >>> compressed_sentences, stats = compressor.compress(sentences, embeddings)
    >>> print(f"Compression: {stats['original_count']} → {stats['compressed_count']} "
    ...       f"({stats['compression_ratio']:.1%} reduction)")
    """

    # Domain-specific aggressiveness presets (empirically tuned)
    # Higher values = more aggressive compression (fewer output sentences)
    _MODE_AGGRESSIVENESS: Dict[str, float] = {
        "legal": 0.25,    # Conservative: preserve legal nuance
        "finance": 0.30,  # Moderate: balance precision and brevity
        "code": 0.45,     # Aggressive: code is repetitive by nature
        "general": 0.40,  # Balanced default
    }

    # Default configuration values
    _DEFAULT_AGGRESSIVENESS: float = 0.25
    _DEFAULT_MIN_VARIANCE: float = 0.0
    _DEFAULT_AUTO_METHOD_THRESHOLD: int = 200

    def __init__(
        self,
        aggressiveness: float = _DEFAULT_AGGRESSIVENESS,
        min_variance: float = _DEFAULT_MIN_VARIANCE,
        mode: Optional[str] = None,
        method: Optional[str] = None,
        auto_method_threshold: int = _DEFAULT_AUTO_METHOD_THRESHOLD,
    ) -> None:
        """
        Initialize the SemanticCompressor.

        Parameters
        ----------
        aggressiveness : float, optional
            Target compression intensity in [0, 1]. Higher = more compression.
            Overridden by mode preset if mode is specified.
        min_variance : float, optional
            Minimum embedding variance to retain a sentence. Sentences with
            variance below this threshold are discarded as low-information.
        mode : Optional[str], optional
            Domain mode: "legal", "finance", "code", or "general".
            Sets aggressiveness preset and may influence future extensions.
        method : Optional[str], optional
            Clustering algorithm: "ward" (hierarchical) or "kmeans" (partition).
            If None, auto-selects based on sentence count vs auto_method_threshold.
        auto_method_threshold : int, optional
            Sentence count threshold for auto-selecting clustering method.
            Below: use Ward (precise). Above: use K-Means (scalable).

        Raises
        ------
        ValueError
            If aggressiveness not in [0, 1] or min_variance < 0.

        Complexity
        ----------
        Time: O(1) initialization
        Space: O(1) additional state
        """
        # Validate parameters
        if not 0.0 <= aggressiveness <= 1.0:
            raise ValueError(f"aggressiveness must be in [0, 1], got {aggressiveness}")
        if min_variance < 0:
            raise ValueError(f"min_variance must be >= 0, got {min_variance}")

        # Store base configuration
        self._base_aggressiveness = aggressiveness
        self.min_variance = min_variance
        self.mode = mode
        self.method = method
        self.auto_method_threshold = auto_method_threshold

        # Resolve effective aggressiveness (mode preset overrides explicit value)
        if mode is not None and mode in self._MODE_AGGRESSIVENESS:
            self.aggressiveness = self._MODE_AGGRESSIVENESS[mode]
            logger.debug(f"Mode '{mode}' preset: aggressiveness={self.aggressiveness}")
        else:
            self.aggressiveness = aggressiveness

        logger.info(
            f"SemanticCompressor initialized: aggressiveness={self.aggressiveness:.2f}, "
            f"min_variance={self.min_variance}, mode={mode or 'manual'}, "
            f"method={method or 'auto'}, auto_threshold={auto_method_threshold}"
        )

    def _remove_low_variance(
        self,
        sentences: List[str],
        embeddings: np.ndarray,
    ) -> Tuple[List[str], np.ndarray, List[int]]:
        """
        Filter out sentences with low embedding variance (low information content).

        Sentences whose embeddings have low variance across dimensions are likely
        generic, repetitive, or semantically empty. Removing them improves
        compression quality by focusing on informative content.

        Parameters
        ----------
        sentences : List[str]
            List of sentence strings.
        embeddings : np.ndarray
            Array of shape (n_sentences, embedding_dim) with sentence embeddings.

        Returns
        -------
        Tuple[List[str], np.ndarray, List[int]]
            - Filtered sentences list
            - Filtered embeddings array
            - Original indices of kept sentences (for order preservation)

        Mathematical Formulation
        ------------------------
        For each embedding eᵢ ∈ ℝᵈ:
            varianceᵢ = (1/d) Σⱼ (eᵢⱼ - μᵢ)²
            where μᵢ = (1/d) Σⱼ eᵢⱼ (mean of embedding components)
        
        Keep sentence i iff: varianceᵢ >= min_variance

        Complexity
        ----------
        Time: O(n·d) where n=sentences, d=embedding_dim
        Space: O(n) for variance array + output lists

        Reference
        ---------
        [1] Jolliffe, I. T. (2002). Principal Component Analysis.
        """
        if len(sentences) == 0:
            return [], np.array([]).reshape(0, embeddings.shape[1] if embeddings.ndim > 1 else 0), []

        # Compute variance across embedding dimensions for each sentence
        variances = np.var(embeddings, axis=1)
        
        # Boolean mask: keep sentences with variance >= threshold
        mask = variances >= self.min_variance
        kept_indices = np.where(mask)[0].tolist()
        
        discarded_count = len(sentences) - len(kept_indices)
        if discarded_count > 0:
            logger.info(f"Filtered {discarded_count} low-variance sentences "
                        f"(variance < {self.min_variance:.4f})")
        
        # Return filtered data with original indices for order tracking
        return (
            [sentences[i] for i in kept_indices],
            embeddings[kept_indices],
            kept_indices,
        )


    def _compute_condensed_distance(self, embeddings: np.ndarray) -> np.ndarray:
        """
        Compute condensed distance matrix for hierarchical clustering.

        Converts square distance matrix to condensed form (upper triangle only)
        as required by scipy.cluster.hierarchy.linkage().

        Parameters
        ----------
        embeddings : np.ndarray
            Array of shape (n, d) with sentence embeddings.

        Returns
        -------
        np.ndarray
            Condensed distance matrix of length n*(n-1)/2.

        Mathematical Note
        -----------------
        Input: Square distance matrix D ∈ ℝⁿˣⁿ where Dᵢⱼ = ||eᵢ - eⱼ||₂
        Output: Condensed vector containing Dᵢⱼ for all i < j
        
        Length of output: n·(n-1)/2 (number of unique pairs)

        Complexity
        ----------
        Time: O(n²·d) for pairwise Euclidean distances via cdist
        Space: O(n²) for full distance matrix, O(n²/2) for condensed output

        Reference
        ---------
        [2] scipy.spatial.distance.squareform documentation
        """
        # Compute full pairwise Euclidean distance matrix
        dist_square = cdist(embeddings, embeddings, metric="euclidean")
        
        # Zero diagonal (distance to self) to avoid numerical issues
        np.fill_diagonal(dist_square, 0)
        
        # Convert to condensed form (upper triangle, flattened)
        return squareform(dist_square, checks=True)


    def _compute_percentile_cutoff(
        self, 
        dist_square: np.ndarray, 
        aggressiveness: float
    ) -> float:
        """
        Compute adaptive distance cutoff using percentile of empirical distribution.

        The cutoff determines cluster merging threshold in Ward clustering:
        clusters with inter-cluster distance < cutoff are merged.

        Parameters
        ----------
        dist_square : np.ndarray
            Square pairwise distance matrix (n x n).
        aggressiveness : float
            Compression intensity in [0, 1]. Lower = stricter merging.

        Returns
        -------
        float
            Distance threshold for cluster cutting.

        Mathematical Formulation
        ------------------------
        Let U = {Dᵢⱼ : 0 <= i < j < n} be upper-triangular distances.
        cutoff = percentile(U, q = aggressiveness * 100)
        
        Interpretation:
        - aggressiveness=0.0 → cutoff=min(U) → merge only identical → no compression
        - aggressiveness=1.0 → cutoff=max(U) → merge everything → max compression
        - aggressiveness=0.3 → cutoff=30th percentile → moderate merging

        Complexity
        ----------
        Time: O(n²) to extract upper triangle + O(n² log n) for percentile
        Space: O(n²) for temporary distance array

        Note
        ----
        For large n, consider sampling distances for approximate percentile.
        """
        # Extract upper triangle (unique pairwise distances, exclude diagonal)
        triu_indices = np.triu_indices_from(dist_square, k=1)
        all_distances = dist_square[triu_indices]
        
        if len(all_distances) == 0:
            return 0.0
        
        # Compute percentile cutoff
        percentile_q = aggressiveness * 100
        cutoff = float(np.percentile(all_distances, percentile_q))
        
        logger.debug(
            f"Percentile cutoff: P({percentile_q:.1f}) = {cutoff:.4f} "
            f"(range: [{all_distances.min():.4f}, {all_distances.max():.4f}])"
        )
        return cutoff


    def _cluster_kmeans(
        self,
        sentences: List[str],
        embeddings: np.ndarray,
        aggressiveness: float,
        original_indices: List[int],
    ) -> Tuple[List[str], List[int]]:
        """
        Cluster sentences using MiniBatch K-Means and select centroid representatives.

        MiniBatch K-Means provides O(n) scalability vs. O(n²) for standard K-Means,
        making it suitable for large sentence sets while maintaining quality.

        Parameters
        ----------
        sentences : List[str]
            List of sentence strings to cluster.
        embeddings : np.ndarray
            Array of shape (n, d) with sentence embeddings.
        aggressiveness : float
            Compression intensity: determines number of clusters as
            n_clusters = max(1, n * (1 - aggressiveness)).
        original_indices : List[int]
            Original positions of sentences for order preservation.

        Returns
        -------
        Tuple[List[str], List[int]]
            - Selected representative sentences (one per cluster)
            - Their original indices (sorted for order preservation)

        Algorithm
        ---------
        1. Compute target clusters: k = max(1, n * (1 - aggressiveness))
        2. Fit MiniBatchKMeans with k clusters
        3. For each cluster:
            a. Compute centroid as mean of member embeddings
            b. Select sentence closest to centroid (most representative)
        4. Sort selected sentences by original index

        Complexity
        ----------
        Time: O(n·k·d·i) where i=iterations (typically 10-100)
        Space: O(n·d + k·d) for embeddings + centroids

        Reference
        ---------
        [4] Sculley, D. (2010). Web-Scale K-Means Clustering.
        """
        n = len(sentences)
        if n == 0:
            return [], []
        
        # Compute target number of clusters based on aggressiveness
        n_clusters = max(1, int(n * (1.0 - aggressiveness)))
        logger.debug(
            f"K-Means clustering: n={n}, aggressiveness={aggressiveness:.2f} → "
            f"n_clusters={n_clusters}"
        )

        # Fit MiniBatch K-Means (efficient for large n)
        # Normalize embeddings shape: ensure 2D (n_samples, n_features)
        if embeddings.ndim == 3:
            # Common pattern: (n, 1, d) → squeeze singleton middle dim
            if embeddings.shape[1] == 1:
                embeddings = embeddings.reshape(embeddings.shape[0], embeddings.shape[2])
                logger.debug("Squeezed embeddings from 3D to 2D for KMeans")
            else:
                embeddings = embeddings.reshape(embeddings.shape[0], -1)
                logger.warning(
                    "Flattened 3D embeddings to 2D for KMeans; verify embedding generation"
                )
        elif embeddings.ndim != 2:
            raise ValueError(f"Embeddings must be 2D array, got ndim={embeddings.ndim}")

        kmeans = MiniBatchKMeans(
            n_clusters=n_clusters,
            random_state=42,  # Reproducibility
            n_init="auto",  # Use auto initialization for modern sklearn
            max_iter=300,
            batch_size=min(100, n),  # Adaptive batch size
        )
        labels = kmeans.fit_predict(embeddings)

        selected_sentences: List[str] = []
        selected_indices: List[int] = []

        # Select most central sentence from each cluster
        for cluster_id in np.unique(labels):
            # Indices of sentences in this cluster
            cluster_mask = labels == cluster_id
            cluster_indices = np.where(cluster_mask)[0]
            
            if len(cluster_indices) == 1:
                # Singleton cluster: keep the only sentence
                rel_idx = cluster_indices[0]
                selected_sentences.append(sentences[rel_idx])
                selected_indices.append(original_indices[rel_idx])
                continue
            
            # Compute centroid and find closest sentence
            cluster_embeddings = embeddings[cluster_indices]
            centroid = np.mean(cluster_embeddings, axis=0, keepdims=True)
            
            # Euclidean distance from each member to centroid
            distances_to_centroid = cdist(
                cluster_embeddings, centroid, metric="euclidean"
            ).flatten()
            
            # Select sentence with minimum distance to centroid
            best_rel_idx = cluster_indices[np.argmin(distances_to_centroid)]
            selected_sentences.append(sentences[best_rel_idx])
            selected_indices.append(original_indices[best_rel_idx])

        # Sort by original index to preserve document order
        order = np.argsort(selected_indices)
        return (
            [selected_sentences[i] for i in order],
            [selected_indices[i] for i in order],
        )


    def _cluster_ward(
        self,
        sentences: List[str],
        embeddings: np.ndarray,
        aggressiveness: float,
        original_indices: List[int],
    ) -> Tuple[List[str], List[int]]:
        """
        Cluster sentences using Ward's hierarchical method and select representatives.

        Ward's method minimizes within-cluster variance at each merge step,
        producing high-quality clusters for semantic grouping. Best for small n.

        Parameters
        ----------
        sentences : List[str]
            List of sentence strings to cluster.
        embeddings : np.ndarray
            Array of shape (n, d) with sentence embeddings.
        aggressiveness : float
            Compression intensity: determines distance cutoff for cluster cutting.
        original_indices : List[int]
            Original positions of sentences for order preservation.

        Returns
        -------
        Tuple[List[str], List[int]]
            - Selected representative sentences (one per cluster)
            - Their original indices (sorted for order preservation)

        Algorithm
        ---------
        1. Compute pairwise Euclidean distance matrix
        2. Build linkage tree using Ward's variance-minimizing criterion
        3. Cut tree at adaptive distance threshold (percentile-based)
        4. For each cluster: select sentence closest to cluster centroid
        5. Sort selected sentences by original index

        Complexity
        ----------
        Time: O(n²·d) for distances + O(n² log n) for linkage + O(n·k·d) for selection
        Space: O(n²) for distance matrix + O(n) for linkage tree

        Reference
        ---------
        [3] Ward, J. H. (1963). Hierarchical Grouping to Optimize an Objective Function.
        """
        n = len(sentences)
        if n <= 1:
            return sentences.copy(), original_indices.copy()

        # Compute pairwise Euclidean distances
        # Ensure embeddings are 2D
        if embeddings.ndim == 3:
            if embeddings.shape[1] == 1:
                embeddings = embeddings.reshape(embeddings.shape[0], embeddings.shape[2])
                logger.debug("Squeezed embeddings from 3D to 2D for Ward clustering")
            else:
                embeddings = embeddings.reshape(embeddings.shape[0], -1)
                logger.warning(
                    "Flattened 3D embeddings to 2D for Ward clustering; verify embedding generation"
                )
        elif embeddings.ndim != 2:
            raise ValueError(f"Embeddings must be 2D array, got ndim={embeddings.ndim}")

        dist_square = cdist(embeddings, embeddings, metric="euclidean")
        np.fill_diagonal(dist_square, 0)

        # Convert to condensed form for scipy linkage
        condensed_dist = self._compute_condensed_distance(embeddings)
        
        # Build hierarchical clustering tree (Ward's method)
        linkage_matrix = linkage(condensed_dist, method="ward")

        # Compute adaptive cutoff based on aggressiveness
        cutoff = self._compute_percentile_cutoff(dist_square, aggressiveness)
        
        # Cut tree to form flat clusters
        labels = fcluster(linkage_matrix, t=cutoff, criterion="distance")
        n_clusters = len(np.unique(labels))
        logger.debug(
            f"Ward clustering: n={n}, cutoff={cutoff:.4f} → {n_clusters} clusters"
        )

        selected_sentences: List[str] = []
        selected_indices: List[int] = []

        # Select most central sentence from each cluster
        for cluster_id in np.unique(labels):
            cluster_mask = labels == cluster_id
            cluster_indices = np.where(cluster_mask)[0]
            
            if len(cluster_indices) == 1:
                rel_idx = cluster_indices[0]
                selected_sentences.append(sentences[rel_idx])
                selected_indices.append(original_indices[rel_idx])
                continue
            
            # Compute centroid and find closest sentence
            cluster_embeddings = embeddings[cluster_indices]
            centroid = np.mean(cluster_embeddings, axis=0, keepdims=True)
            distances = cdist(cluster_embeddings, centroid, metric="euclidean").flatten()
            
            best_rel_idx = cluster_indices[np.argmin(distances)]
            selected_sentences.append(sentences[best_rel_idx])
            selected_indices.append(original_indices[best_rel_idx])

        # Sort by original index to preserve document order
        order = np.argsort(selected_indices)
        return (
            [selected_sentences[i] for i in order],
            [selected_indices[i] for i in order],
        )


    def compress(
        self,
        sentences: List[str],
        embeddings: np.ndarray,
        aggressiveness: Optional[float] = None,
        mode: Optional[str] = None,
    ) -> Tuple[List[str], Dict[str, any]]:
        """
        Compress sentences by clustering semantically similar ones.

        Main entry point for semantic compression. Preserves protected placeholders
        (__PROT_*) and applies adaptive clustering to compressible content.

        Parameters
        ----------
        sentences : List[str]
            List of sentences to compress (may include protected placeholders).
        embeddings : np.ndarray
            Array of shape (len(sentences), embedding_dim) with precomputed embeddings.
        aggressiveness : Optional[float], optional
            Override compression intensity for this call. If None, uses instance default.
        mode : Optional[str], optional
            Override domain mode for this call. Affects aggressiveness preset.

        Returns
        -------
        Tuple[List[str], Dict[str, any]]
            - Compressed list of sentences (protected + representatives)
            - Statistics dictionary with:
                * original_count: input sentence count
                * compressed_count: output sentence count
                * compression_ratio: 1 - (compressed/original)
                * discarded_low_variance: count removed by variance filter
                * aggressiveness_used: effective aggressiveness value
                * duration_seconds: processing time
                * cluster_method: "ward" or "kmeans"
                * compressed_indices: original indices of kept sentences

        Raises
        ------
        ValueError
            If len(sentences) != embeddings.shape[0].

        Pipeline Overview
        -----------------
        1. Validate inputs and resolve effective aggressiveness
        2. Separate protected (__PROT_*) from compressible sentences
        3. Filter low-variance embeddings (noise removal)
        4. Auto-select clustering method based on sentence count
        5. Cluster and select representatives
        6. Merge protected + compressed, preserving original order
        7. Compute and return statistics

        Complexity
        ----------
        Overall: 
          * Small n (<200): O(n²·d) dominated by Ward clustering
          * Large n (≥200): O(n·k·d·i) dominated by K-Means
        where n=sentences, d=embedding_dim, k=clusters, i=iterations

        Space: O(n²) for Ward distance matrix, O(n·d) for K-Means

        Example
        -------
        >>> compressor = SemanticCompressor(mode="code")
        >>> compressed, stats = compressor.compress(sentences, embeddings)
        >>> print(f"Reduced {stats['original_count']} → {stats['compressed_count']} "
        ...       f"({stats['compression_ratio']:.1%} savings)")
        """
        # Validate input dimensions
        if len(sentences) != embeddings.shape[0]:
            raise ValueError(
                f"Mismatch: {len(sentences)} sentences vs "
                f"{embeddings.shape[0]} embeddings"
            )
        if not sentences:
            return [], {
                "original_count": 0,
                "compressed_count": 0,
                "compression_ratio": 0.0,
            }

        # Resolve effective aggressiveness (call-time override > mode preset > instance default)
        if aggressiveness is not None:
            effective_agg = aggressiveness
        elif mode is not None and mode in self._MODE_AGGRESSIVENESS:
            effective_agg = self._MODE_AGGRESSIVENESS[mode]
        else:
            effective_agg = self.aggressiveness

        logger.debug(f"Effective aggressiveness: {effective_agg:.2f}")

        start_time = time.time()
        original_count = len(sentences)

        # Step 0: Separate protected placeholders from compressible content
        protected_sentences: List[str] = []
        protected_indices: List[int] = []
        normal_sentences: List[str] = []
        normal_embeddings: List[np.ndarray] = []
        normal_original_indices: List[int] = []

        for i, sent in enumerate(sentences):
            if sent.startswith("__PROT_"):
                # Protected content: never compress, always preserve
                protected_sentences.append(sent)
                protected_indices.append(i)
            else:
                normal_sentences.append(sent)
                normal_embeddings.append(embeddings[i])
                normal_original_indices.append(i)

        # Edge case: all content is protected → no compression possible
        if not normal_sentences:
            return sentences, {
                "original_count": original_count,
                "compressed_count": original_count,
                "compression_ratio": 0.0,
                "discarded_low_variance": 0,
                "aggressiveness_used": effective_agg,
                "duration_seconds": time.time() - start_time,
            }

        # Step 1: Filter low-variance (low-information) sentences
        normal_embeddings_arr = np.array(normal_embeddings)
        filtered_sentences, filtered_embeddings, kept_local_indices = self._remove_low_variance(
            normal_sentences, normal_embeddings_arr
        )
        filtered_original_indices = [normal_original_indices[i] for i in kept_local_indices]

        # Edge case: all normal sentences filtered out
        if not filtered_sentences:
            result_sentences = protected_sentences
            return result_sentences, {
                "original_count": original_count,
                "compressed_count": len(result_sentences),
                "compression_ratio": 1.0 - (len(result_sentences) / original_count),
                "discarded_low_variance": len(normal_sentences),
                "aggressiveness_used": effective_agg,
                "duration_seconds": time.time() - start_time,
            }

        # Step 2: Auto-select clustering method based on sentence count
        clustering_method = self.method
        if clustering_method is None:
            if len(filtered_sentences) >= self.auto_method_threshold:
                clustering_method = "kmeans"
                logger.debug(f"Auto-selected K-Means (n={len(filtered_sentences)} >= threshold)")
            else:
                clustering_method = "ward"
                logger.debug(f"Auto-selected Ward (n={len(filtered_sentences)} < threshold)")

        logger.info(f"Clustering method: {clustering_method} ({len(filtered_sentences)} sentences)")

        # Step 3: Perform clustering and select representatives
        if clustering_method == "kmeans":
            compressed_normal, compressed_indices = self._cluster_kmeans(
                filtered_sentences,
                filtered_embeddings,
                effective_agg,
                filtered_original_indices,
            )
        else:  # ward
            compressed_normal, compressed_indices = self._cluster_ward(
                filtered_sentences,
                filtered_embeddings,
                effective_agg,
                filtered_original_indices,
            )

        # Step 4: Merge protected + compressed, preserving original order
        # Use index mapping to handle potential collisions
        final_map: Dict[int, str] = {}
        
        # First, add protected sentences at their original positions
        for idx, sent in zip(protected_indices, protected_sentences):
            final_map[idx] = sent
        
        # Then, add compressed sentences, shifting index if collision
        for sent, idx in zip(compressed_normal, compressed_indices):
            while idx in final_map:
                idx += 1  # Linear probing for next available slot
            final_map[idx] = sent

        # Sort by index and extract final sentence list
        sorted_indices = sorted(final_map.keys())
        compressed_sentences = [final_map[i] for i in sorted_indices]

        # Step 5: Compute statistics
        compressed_count = len(compressed_sentences)
        compression_ratio = 1.0 - (compressed_count / original_count) if original_count > 0 else 0.0
        discarded_by_variance = original_count - len(filtered_sentences) - len(protected_sentences)

        stats = {
            "original_count": original_count,
            "compressed_count": compressed_count,
            "compression_ratio": compression_ratio,
            "discarded_low_variance": discarded_by_variance,
            "aggressiveness_used": effective_agg,
            "duration_seconds": time.time() - start_time,
            "cluster_method": clustering_method,
            "compressed_indices": sorted_indices,  # Original indices of kept sentences
        }

        logger.info(
            f"Compression complete: {original_count} → {compressed_count} "
            f"({compression_ratio:.1%} reduction) in {stats['duration_seconds']:.3f}s"
        )
        return compressed_sentences, stats


    async def compress_async(
        self,
        sentences: List[str],
        embeddings: np.ndarray,
        aggressiveness: Optional[float] = None,
        mode: Optional[str] = None,
    ) -> Tuple[List[str], Dict[str, any]]:
        """
        Asynchronous version of compress (non-blocking event loop).

        Offloads CPU-bound compression to a worker thread via asyncio.to_thread,
        preventing event loop starvation in async applications.

        Parameters
        ----------
        sentences : List[str]
            List of sentences to compress.
        embeddings : np.ndarray
            Precomputed embeddings array.
        aggressiveness : Optional[float], optional
            Override compression intensity.
        mode : Optional[str], optional
            Override domain mode.

        Returns
        -------
        Tuple[List[str], Dict[str, any]]
            Compressed sentences and statistics (same as compress()).

        Note
        ----
        - Does not provide true parallelism; uses thread pool for offloading
        - Suitable for high-concurrency async servers (FastAPI, etc.)
        - For true parallelism, use multiprocessing or distributed processing
        """
        return await asyncio.to_thread(
            self.compress, sentences, embeddings, aggressiveness, mode
        )