perf(core): run visual participants (actor, scene) in parallel within a shared decode stream

[Mahnoor-Zaffar]

Summary

Runs the actor and scene visual participants in parallel over the shared decode stream: one decode pass (already shared before this change), per-participant bounded queues, and sentinel-based shutdown. Previously participants processed the decoded frames sequentially; now each participant consumes the shared sampled-frame stream concurrently. Actor remains in-order; backpressure stalls decode when a participant falls behind. Also fixes a deadlock where a participant failing mid-stream would hang the pipeline — the participant error now propagates instead, and the worker threads carry the owning workflow's context so the production progress/cancellation callbacks work from them.

Internal-only; no public API or storage changes.

Motivation

Refs grayhatdevelopers#96. The visual participants (scene, actor) already shared a single decode pass over the video; the cost is that they processed the decoded frames one after the other, so the scene participant's model work serialized with the actor's. This change feeds the shared sampled-frame stream to both participants concurrently so the slower participant overlaps with the faster one instead of waiting for it.

What changed

• src/vidxp/capabilities/visual.py — parallel participant scheduling over the shared decode stream:
  • one decode loop produces frames into per-participant bounded queues;
  • scene and actor consumers run concurrently in their own workers;
  • sentinel shutdown terminates consumers cleanly when decode completes or is cancelled;
  • actor remains in-order while scene may lag;
  • backpressure: if a participant falls behind its queue limit, decode stalls so memory stays bounded;
  • mid-stream participant failures are surfaced as errors instead of deadlocking the pipeline (they are joined and re-raised, not swallowed);
  • each worker runs inside its own copy of the owning thread's context so the DBOS-backed progress and cancellation callbacks work from worker threads.
• tests/test_visual_threading.py — 8 unit tests covering concurrent processing, cancellation, decode failures, participant failures, ordering, backpressure/shutdown, and a production-style callback test that asserts progress and cancellation callbacks see the owning thread's context.

Design details

• One shared decode pass: frames advance once per video; participants subscribe to the same sampled-frame stream rather than each running its own pass.
• Bounded queues: each participant has a finite queue depth, so a slow consumer cannot grow memory unboundedly; when the queue is full, the decode loop pauses (backpressure).
• Deterministic shutdown: a sentinel object marks end-of-stream; consumers drain remaining frames then exit. Cancellation and error paths use the same sentinel protocol so no worker leaks.
• Failure propagation: a failing participant's exception is collected during join and raised from the indexing run — reproducing the previously-hanging mid-stream failure now raises the participant error.
• Context propagation: each worker thread runs inside a contextvars.Context copied from the owning thread, so the DBOS workflow ContextVar is visible to the progress (DBOS.set_event) and cancellation (DBOS.workflow_id) calls the workers make.

Validation

• uv run --no-sync python -m pytest -q tests/test_visual_threading.py — 8 passed.
• uv run --no-sync python -m pytest -q tests/test_runner.py tests/test_manifest.py tests/test_generation_manifest.py — passed (reported 27 in the fork PR).
• uv run --no-sync ruff check src/vidxp/capabilities/visual.py tests/test_visual_threading.py — clean.
• Reproduced the mid-stream participant failure that previously deadlocked; it now raises the participant error.
• Verified the context-propagation test fails without the context wiring and passes with it.

Notes

Internal-only performance/reliability change. No public API, storage schema, CLI, or MCP surface changes. Before/after latency and memory measurements for scene alone and scene,actor are pending a reproducible benchmark, per the discussion in grayhatdevelopers#96.

[tulayha]

Thanks for taking this on. The shared stream implementation and queue/shutdown tests are a useful foundation.

I think the PR description needs one clarification: the existing implementation already decoded a single shared frame stream for all visual participants, so this change parallelizes participant processing rather than removing multiple decode passes.

Before merging, could we also add before/after latency and memory measurements for both scene alone and scene,actor? The discussion on issue #96 placed the benchmark first, and the scene-only case would help show the overhead of the new thread and queues when only one participant is selected.

[tulayha]

One production path may need adjustment here. These worker threads can call the progress and cancellation callbacks outside the thread that owns the DBOS workflow context. The production progress callback uses DBOS.set_event(), while DBOS keeps the active workflow state in a ContextVar that raw threading.Threads do not inherit. This means the first progress update may fail the indexing job. The current tests do not exercise that path because they use progress=None and a plain cancellation token. Could we keep these callbacks on the owning thread, or carry the required context into the workers, and add a production-style callback test?

[Mahnoor-Zaffar]

on it

[Mahnoor-Zaffar]

Addressed in 43d4541: each worker thread now runs inside its own contextvars.Context copied from the owning thread (_in_context in _consume_visual_stream), so the DBOS workflow ContextVar is visible to the progress (DBOS.set_event) and cancellation (DBOS.workflow_id) calls made from the decode and participant workers. Each worker gets a separate copy of the context, since a single Context cannot be entered from more than one thread at a time.

Added test_worker_callbacks_inherit_owning_thread_context in tests/test_visual_threading.py: it wires a ContextVar-gated progress callback (like DBOS.set_event) and a context-dependent cancellation event, then runs _consume_visual_stream with them. Verified the test fails without the context wiring and passes with it; tests/test_visual_threading.py now 8 passed.