Add context manager (__enter__ / __exit__) support to InferencePipeline

Title

Add context manager (__enter__ / __exit__) support to InferencePipeline

Description

Search before asking

• I searched the existing issues and found no similar request for context manager support on InferencePipeline.

Feature Description

Add the Python context manager protocol (__enter__ / __exit__) to InferencePipeline so it can be used with with statements, guaranteeing resource cleanup even when exceptions occur.

Motivation — the resource leak problem

Currently, the only way to run an InferencePipeline is the manual lifecycle pattern:

pipeline = InferencePipeline.init(...)
pipeline.start()
pipeline.join()

If any exception is raised between start() and join() — whether from user callback code, an inference error, or a sink failure — join() is never reached. This leaks resources because join() is responsible for:

1. Joining the inference thread (self._inference_thread.join()) — line 938
2. Joining the dispatching thread (self._dispatching_thread.join()) — line 941
3. Calling on_pipeline_end (line 943), which:
   • Shuts down the internal ThreadPoolExecutor (thread_pool_executor.shutdown()) — inference/core/interfaces/stream/utils.py:128
   • Saves workflow profiling traces to disk — utils.py:117

Without join(), orphan threads and thread pool workers keep running in the background, and profiling data is silently lost.

terminate() alone is not sufficient — it signals threads to stop (self._stop = True) and purges video source buffers, but it does not join threads or invoke on_pipeline_end. Both terminate() and join() must be called for a clean shutdown.

Precedent in the same codebase

VideoFileSink in inference/core/interfaces/stream/sinks.py (lines 537–541) already implements this pattern:

def __enter__(self) -> "VideoFileSink":
    return self

def __exit__(self, exc_type, exc_val, exc_tb) -> None:
    self.release()

Adding the same protocol to InferencePipeline would make the API consistent across the streaming subsystem.

Proposed Implementation

Add two methods to InferencePipeline:

def __enter__(self) -> "InferencePipeline":
    return self

def __exit__(self, exc_type, exc_val, exc_tb) -> None:
    self.terminate()
    self.join()

The __exit__ method calls terminate() first to signal all threads to stop (sets _stop = True, purges video source buffers), then join() to wait for thread completion and run the on_pipeline_end cleanup (thread pool shutdown, profiling save). The exception parameters are intentionally not consumed — they are re-raised automatically by the with statement after cleanup.

Example Usage

Before (current — resource leak on exception):

pipeline = InferencePipeline.init(
    video_reference="./video.mp4",
    model_id="my-model/1",
    on_prediction=my_sink,
)
try:
    pipeline.start()
except Exception:
    # If start() partially succeeded (inference thread launched),
    # we leak that thread. Even if we catch the error, join() is
    # not called unless we explicitly handle it here.
    pipeline.terminate()
    pipeline.join()
    raise
pipeline.join()

After (with context manager — guaranteed cleanup):

with InferencePipeline.init(
    video_reference="./video.mp4",
    model_id="my-model/1",
    on_prediction=my_sink,
) as pipeline:
    pipeline.start()
# terminate() + join() are called automatically, even on exceptions

Alternatives Considered

1. try/finally wrapper — works but is boilerplate-heavy and error-prone; every user must remember to call both terminate() and join() in the correct order. The context manager encapsulates this once.

2. Making start() return a context manager — more invasive; changes the return type of an existing public API and would break callers that use start() without with.

3. Adding cleanup to terminate() — would change the semantics of an existing method. terminate() is designed as a lightweight signal (set _stop, purge buffers); folding thread-joining and profiler-saving into it would be a breaking change for users who call terminate() from a different thread and join() separately.

Additional context

• The change is backward-compatible — existing code using init() → start() → join() continues to work unchanged.
• The implementation is ~8 lines of code with no new dependencies.
• Tests can be added to verify that __exit__ calls terminate() and join() (including the exception path).

Find the InferencePipeline class and inspect its existing start(), terminate(), and join() lifecycle methods, then review the context-manager precedent in inference/core/interfaces/stream/sinks.py. Add focused tests for cleanup on normal and exception paths, verifying that terminate() precedes join() and that exceptions are re-raised.