Fix PipelineTask double-inserting RTVIProcessor when custom RTVIObserver is provided

When the user places an RTVIProcessor inside their pipeline and provides
a custom RTVIObserver subclass in observers, PipelineTask correctly
detects both and logs "skipping default ones." However it then
unconditionally prepends self._rtvi to the pipeline, causing the
processor to appear twice in the frame chain.

Track whether the RTVIProcessor was found externally (inside the user
pipeline) vs created internally. Only prepend it when created internally.

Fixes #3867
This commit is contained in:
Rupesh
2026-02-27 13:29:01 -08:00
parent 83e29eb478
commit 6f33aff0c6
2 changed files with 10 additions and 1 deletions

1
changelog/3867.fixed.md Normal file
View File

@@ -0,0 +1 @@
- Fixed `PipelineTask` double-inserting `RTVIProcessor` into the frame chain when the user provides both an `RTVIProcessor` in the pipeline and a custom `RTVIObserver` subclass in observers.

View File

@@ -330,6 +330,7 @@ class PipelineTask(BasePipelineTask):
# RTVI support # RTVI support
self._rtvi = None self._rtvi = None
self._rtvi_external = False
external_rtvi = self._find_processor(pipeline, RTVIProcessor) external_rtvi = self._find_processor(pipeline, RTVIProcessor)
external_observer_found = any(isinstance(o, RTVIObserver) for o in observers) external_observer_found = any(isinstance(o, RTVIObserver) for o in observers)
@@ -349,6 +350,7 @@ class PipelineTask(BasePipelineTask):
"They are both added by default, no need to add them yourself." "They are both added by default, no need to add them yourself."
) )
self._rtvi = external_rtvi self._rtvi = external_rtvi
self._rtvi_external = True
elif enable_rtvi: elif enable_rtvi:
self._rtvi = rtvi_processor or RTVIProcessor() self._rtvi = rtvi_processor or RTVIProcessor()
observers.append(self._rtvi.create_rtvi_observer(params=rtvi_observer_params)) observers.append(self._rtvi.create_rtvi_observer(params=rtvi_observer_params))
@@ -388,7 +390,13 @@ class PipelineTask(BasePipelineTask):
# allows us to receive and react to downstream frames. # allows us to receive and react to downstream frames.
source = PipelineSource(self._source_push_frame, name=f"{self}::Source") source = PipelineSource(self._source_push_frame, name=f"{self}::Source")
sink = PipelineSink(self._sink_push_frame, name=f"{self}::Sink") sink = PipelineSink(self._sink_push_frame, name=f"{self}::Sink")
processors = [self._rtvi, pipeline] if self._rtvi else [pipeline] # Only prepend the RTVIProcessor if we created it ourselves. When the
# user already placed it inside their pipeline we must not insert it
# again or it will appear twice in the frame chain.
if self._rtvi and not self._rtvi_external:
processors = [self._rtvi, pipeline]
else:
processors = [pipeline]
self._pipeline = Pipeline(processors, source=source, sink=sink) self._pipeline = Pipeline(processors, source=source, sink=sink)
# The task observer acts as a proxy to the provided observers. This way, # The task observer acts as a proxy to the provided observers. This way,