What causes "Pipeline processing failed: {e}" in ruvnet/RuView?

Unwrap the message chain: the text nests stage errors — fix the innermost cause (undersized batch, shape, NaN).

ruvnet/RuView · error · CSIProcessingError

Pipeline processing failed: {e}

Error message

Pipeline processing failed: {e}

What it means

CSIProcessingError raised by the async CSIProcessor.process_csi_data, the end-to-end pipeline (preprocess -> extract -> detect -> add_to_history). Any exception from any stage is re-wrapped under 'Pipeline processing failed'; the handler also increments the internal _processing_errors counter, so repeated failures are visible via processor stats. The nested original message (possibly double-wrapped, e.g. 'Failed to preprocess CSI data: ...') identifies the failing stage.

Source

Thrown at archive/v1/src/core/csi_processor.py:265

            self._total_processed += 1
            
            # Preprocess the data
            preprocessed_data = self.preprocess_csi_data(csi_data)
            
            # Extract features
            features = self.extract_features(preprocessed_data)
            
            # Detect human presence
            detection_result = self.detect_human_presence(features)
            
            # Add to history
            self.add_to_history(csi_data)
            
            return detection_result
            
        except Exception as e:
            self._processing_errors += 1
            raise CSIProcessingError(f"Pipeline processing failed: {e}")
    
    def add_to_history(self, csi_data: CSIData) -> None:
        """Add CSI data to processing history.

        Args:
            csi_data: CSI data to add to history
        """
        self.csi_history.append(csi_data)
        # Cache mean phase for fast Doppler extraction
        if csi_data.phase.ndim == 2:
            self._phase_cache.append(np.mean(csi_data.phase, axis=0))
        else:
            self._phase_cache.append(csi_data.phase.flatten())
    
    def clear_history(self) -> None:
        """Clear the CSI data history."""
        self.csi_history.clear()
        self._phase_cache.clear()

View on GitHub (pinned to 4685618388)

Solutions

Unwrap the message chain: the text nests stage errors — fix the innermost cause (undersized batch, shape, NaN).
Buffer at the source: only call process_csi_data when len(amplitude) >= window_size.
Validate CSIData shape/NaN before entering the pipeline (see validationCode) and drop bad batches with a logged warning.
Give each concurrent consumer its own CSIProcessor; check processor._processing_errors in monitoring to detect chronic bad input.

Example fix

# before
result = await processor.process_csi_data(csi_data)  # first tiny packet raises

# after
if csi_data.amplitude.shape[0] >= processor.window_size:
    result = await processor.process_csi_data(csi_data)
else:
    buffer.append(csi_data)  # wait for more samples

Defensive patterns

Strategy: try-catch

Validate before calling

def pipeline_ready(processor, csi_data) -> bool:
    return (
        csi_data.amplitude.ndim == 2
        and csi_data.amplitude.shape == csi_data.phase.shape
        and csi_data.amplitude.shape[0] >= processor.window_size
        and bool(np.isfinite(csi_data.amplitude).all())
    )

if pipeline_ready(processor, csi_data):
    result = await processor.process_csi_data(csi_data)

Try / catch

try:
    result = await processor.process_csi_data(csi_data)
except CSIProcessingError as e:
    logger.warning('pipeline failure on batch %s: %s',
                   getattr(csi_data, 'timestamp', '?'), e)
    continue  # drop batch; monitor processor._processing_errors rate
# abort (re-raise) when the error rate exceeds a threshold

Prevention

Gate the stream: only await process_csi_data once the buffer holds >= window_size samples.
Drop corrupted batches at ingestion with a logged warning instead of feeding them to the pipeline.
Track _processing_errors / total ratio as a health metric; alert on sustained bad-input rates.
Use one CSIProcessor per coroutine; never share across tasks.

When it happens

Trigger: Awaiting processor.process_csi_data(csi_data) with data that fails any single stage: undersized batches (< window_size), shape mismatches, NaNs, or invalid internal state. Because it catches Exception, programming errors in custom config or data classes also surface here.

Common situations: Streaming loops pushing every ESP32 packet immediately, including the first undersized ones; network packet loss producing ragged final windows; switching hardware profiles mid-stream changing subcarrier counts; concurrency where multiple coroutines share one processor and corrupt history state.

Related errors

AI-assisted analysis of ruvnet/RuView@4685618388 (2026-08-16). Data as JSON: /api/errors/93984e38d229891c. Report an issue: GitHub.