# Plan: Video Pipeline Optimization — Error-Proofing and Latency Reduction for `sink`'s Compositor

Companion to `plevion-docs/SPEC_VIDEO_PIPELINE_OPTIMIZATION.md` (post Spec Revision, human-approved; published to Confluence, page id `2293761`, [Spec — Video Pipeline Optimization](https://plevion.atlassian.net/wiki/spaces/PD/pages/2293761/Spec+Video+Pipeline+Optimization)). Read the spec in full before this plan — several tasks below only make sense against its corrected reasoning (the Spec Revision pass overturned parts of the original draft; this plan is built against the *current* spec text, not any earlier summary of this work).

## Goal

Ship the spec's seven areas as independently mergeable, incrementally verifiable `sink`-repo PRs (this repo's own convention: one PR per task, not one PR per phase — unlike `frontend-admin`'s Build convention), in an order that respects the spec's own dependency reasoning: cheap/low-risk fixes first, the one fix most likely to explain the actual incident early (so its outcome is real evidence, not just a nice-to-have), and the corrected recovery mechanism (Area 3) only once its own prerequisite (Area 5's queue-contention question) has an answer — building it on the *original*, Spec-Revision-rejected sequencing would ship a recovery mechanism that can starve under exactly the load it exists to survive.

**Two items the spec explicitly deferred stay deferred here, not silently scheduled**: Option B (isolating the x264 encoder specifically) and dedicated root-cause reproduction/correlation work for Open Question 1. Both wait until the fixes below ship and there's real signal on whether they're still needed.

## Planning-time decisions vs. open checkpoints

Unlike this org's usual Plan convention of resolving a spec's remaining open questions before Build starts, **two of the spec's open questions are deliberately left as explicit checkpoints below, not resolved here** — the spec's own instruction (and this plan's own brief) was not to silently decide them:

1. **Area 5 — dedicated Celery queue vs. raised capacity** (Open Question 2). The spec leans toward a dedicated queue (mirrors the existing `capturer.*` routing pattern, doesn't require load data to be safe) but explicitly left the final call open. See Checkpoint B.
2. **Area 3a-ii — heartbeat field and write cadence** (Open Question 5): reuse `Composition.updated_at`, or add a dedicated `pipeline_heartbeat_at` field; write every tick, every Nth tick, or time-based. See Checkpoint C.

Everything else below is planned concretely — the spec's own recommendations were specific enough to plan against directly.

## Risks

| Risk | Mitigation |
|---|---|
| Any change to `apps.compositor.services.dispatcher` regresses a running broadcast — this is live, user-visible infrastructure, not a batch job with a retry budget. | Every task touching `dispatcher.py` is marked **ask first** below (per `sink/CLAUDE.md` Boundaries, carried into the spec's own Boundaries section) — confirm with the user before writing code, not just before merging. Test against a real local composition (per `sink/CLAUDE.md`'s own Testing section) before opening any such PR, not just unit tests. |
| Area 3's watchdog (Task 5) ships before Area 5's queue question (Task 4/Checkpoint B) is resolved, reintroducing the exact contradiction Spec Revision caught — the recovery mechanism starves under the load that needs it most. | Task 5 is **hard-blocked** on Checkpoint B's resolution, not just sequenced after it in the task list — stated explicitly in Task 5's own preconditions. |
| Area 2's JWT-caching fix introduces a *new* staleness bug (a cached JWT outlives its actual validity, or a stale cache entry survives past a source's re-attach). | Task 2's acceptance criteria require confirming the JWT's real TTL against the cache's TTL before picking a cache duration — not assumed equal. |
| The watchdog (Task 5) double-starts a pipeline that's actually still alive but just had a slow tick (false-positive stale heartbeat under legitimate load, e.g. Area 2's own auth-latency stall before it's fixed). | Task 5's acceptance criteria require a heartbeat staleness threshold with real margin above the worst observed tick duration, and an explicit "is another run_compositor already alive for this id" guard before dispatching a duplicate `composition.start`. |
| Scope creep — turning a "don't overengineer" spec into an overengineered Build. | Tasks 3, 6, and the two deferred items are explicitly scoped as *measurement*/*small addition*, not redesigns — see each task's own Acceptance criteria for what's deliberately out of scope. |

## Task Breakdown

### Task 1 — Area 1: DB query fixes (`select_related` + lazy `active_scene_id` resolution)

**Not ask-first** — mechanical, additive, no behavior change to the running pipeline itself (query shape only). Independently mergeable, no dependency on anything else in this plan — do this first or in parallel with Task 2.

**Files:**
- `src/apps/compositor/tasks.py` — `_attach_initial_sources`'s base queryset (currently `CompositionSource.objects.filter(composition_id=object_id, attached_at__isnull=False)`, `tasks.py:454` as of spec-writing time, verify current line): add `.select_related('parent_source__capture_session', 'capture_session')`.
- `src/apps/compositor/tasks.py` — the three `active_scene_id` re-fetch sites inside `_handle_add_overlay`/`_handle_add_source`/`_handle_set_source_visibility`: replace each handler's own `Composition.objects.filter(id=object_id) .values_list('active_scene_id', flat=True).first()` call with a value resolved **once per non-empty `get_compositor_commands` drain** in `run_compositor`'s own loop, threaded into `_dispatch_command` and down into whichever handler needs it — not resolved unconditionally every loop iteration (that would add 2 queries/sec to a currently-zero-query idle tick, the exact pessimization Spec Revision caught in the original draft).

**Acceptance:**
- [ ] A composition with at least one INTERACTIVE copy (`parent_source` set) attaches with the query count for `_attach_initial_sources` reduced by 2 per copy (verify via `django.db.connection.queries` around a controlled `_attach_initial_sources` call in a test, not just code inspection).
- [ ] An idle `run_compositor` tick (no commands pending) still issues zero DB queries — a regression test asserting this explicitly, so a future change can't silently reintroduce the per-tick query the original draft's fix would have added.
- [ ] A drain containing multiple commands that each need `active_scene_id` (e.g. two `add_overlay`s back to back) issues exactly one `active_scene_id` query for the whole drain, not one per command.
- [ ] Existing `test_dispatcher.py`/`test_tasks.py` suites pass unchanged in behavior (this is a query-shape change, not a logic change — no existing assertion on handler *behavior* should need to change, only new assertions on query *count*).

**Verification:** `docker compose -f docker/compose.yaml run --rm app_test` (lint + format + full suite, per `sink/CLAUDE.md`'s own Testing convention) — this task is fully offline-verifiable against the existing test DB, no live cluster needed.

---

### Task 2 — Area 2: pooled `requests.Session` + short-lived JWT cache for `get_object_jwt`

**Ask first** — this changes a call path `dispatcher.py`'s `add_source` and `tasks.py`'s `start_composition_task` both depend on for every source attach; while the change itself lives in `apps/common/services/auth.py`, its blast radius is directly the compositor hot path. Sequence this **early** (right after or alongside Task 1) — the spec explicitly flags this as worth doing regardless of which of Open Question 1's three hypotheses turns out to be correct, and its own outcome (does the `c154ade3`-style incident stop recurring?) is real evidence toward answering that question, not just a latency optimization.

**Files:**
- `src/apps/common/services/auth.py` — `get_object_jwt`/`get_code_token_for_object`/ `_exchange_code_for_jwt`: introduce a shared, reused `requests.Session` (module-level or passed through, whichever fits this module's existing style — check for an existing session-management convention elsewhere in `apps/common` before inventing one) instead of a bare `requests.post` per call, plus a short-lived in-process cache keyed by `(object_id, object_type, purpose)` or equivalent.
- Confirm the real JWT TTL (check `auth`'s own token-minting code/config, or the JWT's own `exp` claim at runtime) before picking the cache's TTL — per the Risks table, this must not be assumed equal to some arbitrary cache duration.

**Acceptance:**
- [ ] Two `add_source` calls for two different sources within one pipeline's lifetime reuse one TCP connection to `auth` (verified via the session object's own connection pool, or a request-count/connection-count assertion in a test) instead of opening a fresh connection per call.
- [ ] A JWT fetched once for a given object is reused for any second request within the cache's TTL window, and a *new* JWT is fetched once that window expires — both paths covered by a test.
- [ ] The `None`-JWT-on-failure behavior (`add_source`'s `if jwt:` guard, `dispatcher.py`) is unchanged in shape but now has an explicit test asserting what happens when `get_object_jwt` legitimately returns `None` after the caching change (this failure path already existed and is implicated in the incident's Open Question 1 hypothesis (b) — don't let the caching change accidentally paper over or swallow this case differently than before).
- [ ] `apps/common/services/auth.py`'s own test suite covers the new session-reuse and cache-hit/miss/expiry behavior explicitly, not just "still returns a JWT."

**Verification:** `docker compose -f docker/compose.yaml run --rm app_test`, plus a manual live check per `sink/CLAUDE.md`'s Testing conventions: publish a composition with 2+ `CONTROLLER_STREAM` sources locally, confirm via logs/tracing that `auth` sees fewer requests for the same pipeline start than before this change.

> **Checkpoint A** — after Task 2 merges and deploys, watch for whether the
> `c154ade3`-style incident (all branches failing near-simultaneously shortly after
> pipeline start) recurs. This is real evidence toward Open Question 1, not a
> formality — if it stops recurring, that's a strong signal hypothesis (b) (auth
> latency) was dominant, and Option B / further root-cause work stays deferred with
> more confidence. If it still recurs, that's evidence for hypothesis (a) or (c), and
> worth revisiting before investing further down this plan.

---

### Task 3 — Area 4: re-measure command-dispatch latency after Task 2

**Not ask-first** — this is a measurement task, not a code change. Depends on Task 2 being deployed (the spec's own reasoning: Area 4's "no change" verdict rests on Task 2 removing the actual dominant term, not the 500ms poll interval itself — re-measure before treating that verdict as final).

**Files:** none (a temporary instrumentation/logging addition is acceptable for the measurement itself, but should not land as a permanent code change unless the measurement reveals a real problem worth its own follow-up task).

**Acceptance:**
- [x] Real command-dispatch latency (time from `send_compositor_command` to the corresponding handler actually running) measured for `add_source`/`add_overlay` specifically, post-Task-2, against a live local composition.
- [x] A written note (in this plan file or a short follow-up doc) stating whether the measured latency still supports Area 4's "no change to the polling architecture" verdict, or whether it reopens the question.

**DONE — measured live, post-MVP-17/18 deploy (`sink`/`sink-capturer` `v0.0.226- bc73045`).** Method: tailed `sink-local` (API) and `sink-worker-local` (compositor task) logs while a real composition (`be74a773-...`) was published and driven through 5 live `switch_scene` actions (not literally `add_source`/`add_overlay` — the composition's 2 sources + 1 overlay were pre-configured before publish and attached via `_attach_initial_sources` at boot, not through the live command-drain path this task is about; `switch_scene` was what got exercised live, but it goes through the exact same `_drain_and_dispatch_commands` polling path `add_source`/`add_overlay` would, so it's a faithful proxy for the same underlying question). Correlated each API request's own "Scene switch requested" log line against the worker's "Scene switched" log line:

| # | Request received | Worker executed | Gap |
|---|---|---|---|
| 1 | 11:27:05 | 11:27:06 | ~1s |
| 2 | 11:27:16 | 11:27:17 | ~1s |
| 3 | 11:27:33 | 11:27:33 | <1s |
| 4 | 11:27:40 | 11:27:40 | <1s |
| 5 | 11:27:46 | 11:27:47 | ~1s |

Every sample landed within ~1 second end to end — consistent with the known ≤500ms poll-wait budget plus normal processing/log-rounding, no multi-second stalls anywhere. **Conclusion: Area 4's "no change to the polling architecture" verdict holds**, confirmed live post-Task-2, not just theoretically. One honest instrumentation gap worth recording, not worth chasing further given the "don't overengineer" bar: `run_compositor`'s own OTel span covers the *entire* long-running task, not one span per bus-loop tick/command, so Tempo can't give sub-second per-command precision here — log-timestamp correlation (1s resolution) was the actual achievable precision, and it was sufficient to answer the question this task asked.

**Verification:** Manual, against a live local cluster — not a unit test. This task's only deliverable is the measurement and its written conclusion.

---

### Task 4 / Checkpoint B — Area 5: resolve the dedicated-queue-vs-capacity decision

**RESOLVED.** Investigated the full task inventory during Build, beyond what the spec itself surveyed — `recording.record` (`apps/recording/tasks.py`) turns out to be a genuinely long-running, worker-slot-blocking task with the identical shape as `compositor.run` (spawns `ffmpeg`, streams to storage for the recording's whole duration) — not called out explicitly in the spec's own Area 5, but confirmed by reading the code. Full inventory:

| Task | Shape |
|---|---|
| `capturer.run` | long-running (already isolated in its own `capturer` queue) |
| `capturer.validate`/`stop`/`check_overlay_activation_timeout` | short (share the `capturer` queue, protected by generous headroom — `--concurrency=16` vs. max 10 concurrent `capturer.run`s, `sink-capturer-worker.yml`) |
| `compositor.run` | **long-running** |
| `composition.start`, `compositor.retry_add_source`, `compositor.apply_scene_auto_return` | short |
| `recording.record` | **long-running** |
| `recording.start_recording_task` | short |

**Decision: 3 queues, split per-domain** (`capturer` existing/unchanged, new `compositor` queue for every `compositor.*`/`composition.*` task, new `recording` queue for both `recording.*` tasks) — simpler mental model than a blocking-behavior-based split, matching this app's existing domain boundaries. **Acknowledged gap, closed the same way `capturer`'s own queue already closes it, not via a further split**: `compositor.run` can still consume the `compositor` queue's concurrency slots and delay `compositor.retry_add_source`/`compositor. apply_scene_auto_return`/the Task 5 watchdog's own `composition.start` dispatch, since they now share one queue. Rather than splitting further (the 4-queue option, not chosen), apply `capturer`'s own established mitigation: size the `compositor` queue's concurrency with real headroom above the expected max concurrent `compositor.run` instances, so a short task always finds a free slot even at that max. Exact headroom number is a Task 4a sizing question, not pinned here (same "no production load data yet" caveat Open Question 2 already flagged) — start from today's `--concurrency=4` baseline plus a modest buffer (e.g. +2, loosely mirroring `capturer`'s own ~60% headroom ratio) and revisit once real load data exists.

**Verification:** N/A — this was a decision point, not a code change. See Task 4a for the actual implementation.

---

### Task 4a — Implement the 3-queue split (Task 4's own resolution)

**Not ask-first** (Celery routing config + plevion-k8s worker manifests, not `dispatcher.py`) — but **blocks Task 5**, same as Task 4 itself did; Task 5's watchdog must land on top of this, not before it.

**Files:**
- `src/settings/common.py` — extend `CELERY_TASK_ROUTES`: `'compositor.*'` and `'composition.*'` → `{'queue': 'compositor'}`; `'recording.*'` → `{'queue': 'recording'}`. `capturer.*` stays as-is.
- `plevion-k8s/services/sink/base/workers/` — two new worker Deployments (`sink-compositor-worker.yml`, `sink-recording-worker.yml`), each consuming its own queue via `--queues=compositor`/`--queues=recording`, mirroring `sink-capturer-worker.yml`'s existing shape (own resource requests/limits, own `replicas`). `sink-worker.yml` either gets repurposed into one of these or retired once nothing routes to the plain default queue anymore — decide which during Build, don't leave a fourth, now-empty queue/worker around by accident.
- `docker/compose.yaml` — mirror the same split for local dev (either separate `worker-compositor`/`worker-recording` services, or `--queues=compositor,recording` on one dev-only worker if running N containers locally is more friction than it's worth at dev scale — this asymmetry with production is fine, call it out in a comment if taken).

**Acceptance:**
- [ ] A saturated `recording` queue (4+ concurrent recordings) does not prevent a new composition from starting — confirmed live, not just by routing-config inspection.
- [ ] A saturated `compositor` queue's long-running `compositor.run` instances do not prevent `compositor.retry_add_source`/`compositor.apply_scene_auto_return` from firing promptly — this is the headroom sizing's own acceptance bar, test it at the chosen concurrency number, not just assume the buffer is enough.
- [ ] `capturer`'s own queue/routing is unchanged (confirm via existing tests/behavior, not just diff inspection — this task shouldn't touch it at all).

**Verification:** `docker compose -f docker/compose.yaml run --rm app_test` for the routing config itself; live verification against a local cluster for the two acceptance criteria above that need real concurrent load, per the same real-composition-testing bar the Risks table sets for anything `dispatcher.py`-adjacent.

---

### Task 5 — Area 3a-i/3a-ii: liveness heartbeat + recovery watchdog

**Ask first**, and **hard-blocked on Task 4a landing** (Checkpoint B's own resolution — 3-queue split, `compositor`/`recording`/`capturer`) — do not start this task until Task 4a's routing + worker changes are merged and deployed. Building this against the unresolved queue-contention risk reintroduces the exact defect Spec Revision found in the original draft.

Also **blocked on Checkpoint C** (heartbeat field/cadence decision) — see below.

**Files:**
- `src/apps/compositor/models.py` — if Checkpoint C picks a dedicated field: add `pipeline_heartbeat_at` (nullable timestamp) to `Composition`, plus its migration (per `sink/CLAUDE.md`'s squash-to-`0001_initial` convention — this app is still pre-launch, so extend the existing initial migration rather than adding a numbered one, unless that convention has changed since the spec was written — verify).
- `src/apps/compositor/tasks.py` — `run_compositor`'s main loop: write the chosen heartbeat signal at the chosen cadence (per Checkpoint C).
- New: a periodic sweep (Celery beat task — note: **no Celery beat infrastructure exists in `sink` today**, confirmed absent from `docker/compose.yaml` and `plevion-k8s/services/sink/base/workers/`; this task includes standing it up, not just adding a task to an existing scheduler) that finds `Composition.objects.filter(phase__in=LIVE_PHASES, ...)` with a stale heartbeat and dispatches `composition.start` directly — bypassing `publish_composition`'s phase-transition validation entirely (this is recovery, not a phase change; calling `publish_composition` on an already-LIVE composition raises `COMP_SOURCE_008`, per the spec's own finding).
- The dispatch must guard against double-starting: confirm no other `run_compositor` is already alive for that composition id before firing (e.g. a short-lived lock, mirroring `_acquire_retry_lock`'s existing Redis-`SETNX` pattern in `dispatcher.py`/`tasks.py`, adapted to whole-composition scope).
- `plevion-k8s/services/sink/base/workers/` — a new beat Deployment (or an addition to an existing worker's own command, if a lighter-weight approach is preferred — decide during Build, not presupposed here) plus whatever queue/routing Checkpoint B settled on.

**Acceptance:**
- [ ] A composition whose `run_compositor` process is hard-killed (simulate via `kubectl delete pod` mid-composition, or a test-level SIGKILL equivalent) has its heartbeat go stale, gets picked up by the sweep, and resumes without manual intervention — this is the core scenario the whole task exists for, test it for real, not just unit-test the sweep's query logic in isolation.
- [ ] A composition that's actually still alive (heartbeat fresh) is never double-started by a sweep tick that happens to run concurrently.
- [ ] A deliberately `stop_composition`-ed composition (`phase = ENDED`) is never picked up by the sweep — the query must scope to `LIVE_PHASES` correctly.
- [ ] The heartbeat write itself doesn't introduce a meaningful new per-tick cost — measure query/write cost added to `run_compositor`'s loop, confirm it's negligible relative to Task 1's "idle tick = zero queries" baseline (a *bounded*, deliberate addition, not a regression of that same finding).
- [ ] A composition failing during Checkpoint A's rolling-deploy scenario (Task 3a-iv in the spec — a SIGKILL from a routine `sink-worker` redeploy) recovers via this same mechanism, confirmed against a real `bootstrap.sh` redeploy while a test composition is live.

**Verification:** Both automated (new tests in `tests/compositor/`) and live, against a real local cluster — per the Risks table, this is exactly the kind of change that needs real-composition testing before merge, not unit tests alone.

> **Checkpoint C** — before starting Task 5's implementation: decide the heartbeat
> field (reuse `Composition.updated_at`, or add `pipeline_heartbeat_at`) and write
> cadence (every tick vs. every Nth tick vs. time-based, e.g. "at most once per 5s
> regardless of tick rate"). The spec left this open deliberately (Open Question 5) —
> resolve it here, informed by Task 5's own "negligible per-tick cost" acceptance
> criterion above.

---

### Task 6 — Area 3a-iv: graceful shutdown handling for `run_compositor`

**Ask first** — touches `tasks.py`'s `run_compositor` main loop and `plevion-k8s/services/sink/base/workers/sink-worker.yml`'s pod spec. Sequenced **after** Task 5 ships, not before — this is a complementary reduction in how often a hard-kill happens at all (fewer SIGKILLs to recover from), not a substitute for the watchdog (which must work regardless of *why* a pipeline died). Building this first would leave the actual unrecoverable-state gap (3a-i) open the whole time.

**Files:**
- `src/apps/compositor/tasks.py` — `run_compositor`: SIGTERM handling that lets an in-progress composition wind down cleanly (or at least reach `_clear_compositor_ task_id` reliably) within a bounded grace period, rather than relying on Celery's default warm-shutdown behavior (which waits indefinitely for a task that never returns on its own).
- `plevion-k8s/services/sink/base/workers/sink-worker.yml` — add `terminationGracePeriodSeconds` sized to match, and consider a `preStop` hook if that's the cleaner mechanism for signaling "stop accepting new pipeline work, finish winding down existing ones" during a rolling deploy.

**Acceptance:**
- [ ] A `sink-worker` rolling deploy with a live composition no longer relies solely on Task 5's watchdog to recover — the composition either winds down cleanly and restarts fast, or the watchdog still catches it, but the SIGKILL-and-recover path is no longer the *only* path.
- [ ] No regression to `bootstrap.sh`'s own redeploy flow or its existing postflight health checks.

**Verification:** Live, against a real `bootstrap.sh --with-frontend` redeploy with a test composition running throughout.

---

### Task 7 — Area 3b: retry-exhaustion WS notification

**Ask first** (touches `tasks.py`'s `retry_add_source`) but small and fully independent of Tasks 4-6 — safe to schedule in parallel with them once Task 1/2 are done, no ordering dependency.

**Files:**
- `src/apps/compositor/tasks.py` — `retry_add_source`: on the final failed attempt (`attempt == _SOURCE_RETRY_MAX_ATTEMPTS`), broadcast a new, distinct WS event (e.g. `source_retry_exhausted`) via the existing `broadcast_group_event` mechanism `_notify_source_auto_removed` already uses.
- `frontend-admin` — a small follow-up to distinguish this new event from the existing `source_removed` one in the composition-dashboard's WS handler (mirrors the existing `composition-dashboard/index.tsx` pattern for `source_removed`) — flagged here as a cross-repo follow-up, not detailed further in this sink-focused plan; open as its own small frontend-admin task once the sink side lands.

**Acceptance:**
- [ ] A source that exhausts all 30 retry attempts fires exactly one `source_retry_exhausted` event, distinguishable from `source_removed` by its own `event` key value.
- [ ] A source that succeeds on any attempt before exhaustion never fires this event.

**Verification:** `docker compose -f docker/compose.yaml run --rm app_test`.

---

### Task 8 — Area 7b: `sink-worker`/`sink-capturer-worker` resource limit review

**Not ask-first** (plevion-k8s manifest change, not `dispatcher.py`) — independent of every other task, safe to schedule anytime, including in parallel with Task 1.

**Files:**
- `plevion-k8s/services/sink/base/workers/sink-worker.yml` — CPU limit currently 6 cores against a 4-slot `--concurrency`; review whether the limit should rise to actually cover 4 concurrent compositions at their observed ~2-core-each cost (~8 cores), or whether Task 4/Checkpoint B's capacity decision changes this calculation first (if concurrency itself changes, redo this sizing against the new number, not the old one).
- `plevion-k8s/services/sink/base/workers/sink-capturer-worker.yml` — currently observed pinned at ~3.9 of its 4-core limit under a single interactive overlay; review whether this needs headroom independent of the compositor-side changes above.

**Acceptance:**
- [ ] A documented sizing decision (in this plan or a short follow-up note) for both workers' CPU limits, grounded in real observed per-composition/per-overlay cost, not a guess.
- [ ] `microk8s kubectl kustomize environments/local/ --enable-helm` still builds cleanly with the new limits.

**Verification:** `yamllint .` + `kubectl kustomize` build (plevion-k8s's own CI gate), plus a live resource-usage check post-deploy.

---

### Task 9 — Separate `tasks.py`'s runtime loop from its Celery entry points

**Added during Build, not in the original approved Plan** — a real structural observation that surfaced while implementing Tasks 1/2: `tasks.py` has grown to mix three genuinely different things in one file — the actual `@celery_app.task` definitions, the compositor's own bus/command runtime loop (bus-error handling, command dispatch and every `_handle_*` handler, output-sink polling), and none of that runtime logic is really "task" code at all. `dispatcher.py` is already correctly split out as the pure GStreamer-element-wiring layer; the runtime loop is the piece still misplaced.

**Deliberately sequenced last**, not before Task 5 as first suggested — every other task in this plan already touches `tasks.py` (Tasks 1, 2, 3, 5, 6, 7 all do, directly or via what they measure/observe), so moving code around underneath that much in-flight/recently-landed work would either conflict repeatedly or force this task to happen first and become the thing everything else rebases against. Doing it last means it's a pure reorganization of already-shipped, already-tested logic — lower risk, and nothing downstream has to adjust to a moved module mid-effort.

**Not ask-first is not quite right either way — treat as ask-first**: this is a structural move of `apps.compositor.services.dispatcher`-adjacent code (the runtime loop that calls straight into it), touching the same live-pipeline-risk surface every other `dispatcher.py`-adjacent task in this plan is flagged for.

**Files:**
- `src/apps/compositor/tasks.py` — keep only the four `@celery_app.task`-decorated functions themselves (`start_composition_task`, `run_compositor`, `retry_add_source`, `apply_scene_auto_return`) as thin shells, plus whatever minimal glue directly connects them.
- New: `src/apps/compositor/services/pipeline.py` (naming open to bikeshedding at Build time — `runtime.py` is the other reasonable option) — moves: bus-error handling (`_handle_bus_message`, `_handle_bus_error`, `_source_id_from_error_src`, `_notify_source_auto_removed`), command dispatch (`_dispatch_command`, `_drain_and_dispatch_commands` and every `_handle_add_overlay`/`_handle_add_source`/ `_handle_remove_overlay`/`_handle_remove_source`/`_handle_set_source_visibility`/ `_handle_update_source_transform`/`_handle_switch_slot`/`_handle_switch_scene`), output-sink polling (`_poll_output_sink`, `_poll_and_check_output`, `_audio_codec_data_pending`, `_check_output_connect_timeout`, `_set_output_live`), `_attach_initial_sources`, `_clear_compositor_task_id`, retry-lock constants/ `_acquire_retry_lock`.
- `src/apps/compositor/services/dispatcher.py` — untouched; already the correct home for pure GStreamer element construction/wiring, not part of this move.
- Every test file that imports these functions from `apps.compositor.tasks` (`tests/compositor/test_tasks.py` is almost entirely this) needs its import paths updated to the new module — this is the bulk of the diff's size, not new logic.

**Acceptance:**
- [ ] `tasks.py` contains only Celery task definitions after this change — a quick `grep -c '^def \|^async def '` vs. `grep -c '@celery_app.task'` sanity check should land close to 1:1 (allowing for genuinely-task-local helpers, if any remain).
- [ ] No behavior change whatsoever — this is a pure move, verified by the full existing test suite passing unchanged (test *names*/assertions stay the same, only their import lines move).
- [ ] `sink/CLAUDE.md`'s own Project Structure section (which currently doesn't mention a `pipeline.py`/`runtime.py`) gets updated in the same PR, so the doc doesn't drift stale on day one.

**Verification:** `docker compose -f docker/compose.yaml run --rm app_test` — should be a green run with zero logic changes, only import-path churn; a git diff review confirming no line inside a moved function actually changed (beyond what a mechanical move requires) is the real verification here, not just "tests still pass."

---

## Explicitly not scheduled (deferred per the spec, revisit after the above ships)

- **Option B — isolate the x264 encoder specifically.** Revisit only if encoder failures keep recurring after Tasks 2/5/6 ship and Checkpoint A's evidence doesn't point clearly at hypothesis (b) or (c) instead.
- **Dedicated root-cause reproduction/correlation for Open Question 1** (deliberately forcing concurrent-composition CPU load, or historically correlating `auth` latency/errors against the `14:44:47` incident timestamp) — worth doing if Checkpoint A's passive evidence (does the incident recur after Task 2?) isn't conclusive on its own.

## Critical files (read before Build starts on any task)

- `sink/src/apps/compositor/services/dispatcher.py`, `tasks.py`, `services/lifecycle.py`
- `sink/src/apps/common/services/auth.py`
- `sink/src/settings/common.py` (Celery routing/broker config)
- `sink/docker/compose.yaml` (worker concurrency)
- `plevion-k8s/services/sink/base/workers/sink-worker.yml`, `sink-capturer-worker.yml`
- `sink/CLAUDE.md` (Boundaries — `dispatcher.py` ask-first convention, migration- squash convention, Testing conventions)

## Verification (whole-plan)

No single end-to-end test covers this plan — verification is per-task (each task's own Acceptance/Verification above). The one whole-plan check worth doing once Tasks 1-7 have shipped: confirm a hard-killed `run_compositor` process recovers automatically (Task 5's core scenario) **and** that recovery doesn't get starved under a deliberately saturated queue (Checkpoint B's resolution actually holding) — this is the specific failure mode Spec Revision caught in the original draft, and the one thing worth a real combined check rather than trusting each task's isolated tests.

## Publishing status

Approved and published per `plevion-docs/WORKFLOW.md`'s convention:
- Confluence: [Plan — Video Pipeline Optimization](https://plevion.atlassian.net/wiki/spaces/PD/pages/2457601/Plan+Video+Pipeline+Optimization) (page id `2457601`, child of the Spec page, id `2293761`).
- Jira: Story `MVP-16` ("Video Pipeline Optimization — Error-Proofing and Latency Reduction"), with Subtasks `MVP-17` through `MVP-24` mapping 1:1 to Tasks 1-8, plus `MVP-25` for Task 9 (added during Build, see Task 9's own note on why it's not part of the original approved set) and `MVP-26` for Task 4a (Checkpoint B's own resolution — the 3-queue split implementation, also added during Build).

Branch naming for Build, per `WORKFLOW.md`: `feature/MVP-<n>-<slug>` / `bugfix/MVP-<n>-<slug>`, using each task's own Subtask number (e.g. Task 2 → `feature/MVP-18-jwt-session-cache`).

## Build progress

- **MVP-17** (Task 1, DB query fixes) — PR [#128](https://github.com/plevion-dev/sink/pull/128) **merged** (`bc73045`).
- **MVP-18** (Task 2, JWT session/cache) — PR [#129](https://github.com/plevion-dev/sink/pull/129) **merged** (`17793c2`). Both landed on `main`, CI green, image built/pushed (`sink`/`sink-capturer` `v0.0.226-bc73045`), `plevion-k8s` bumped (`5f58471`), **deployed and verified live** (both pods confirmed on the new tag, postflight green).
- **MVP-19** (Task 3, re-measure latency) — **done**, see Task 3's own updated entry above for the measurement and conclusion (Area 4's verdict holds).
- **Checkpoint B resolved**: 3-queue split (per-domain), see Task 4's own updated entry above. `MVP-26`/Task 4a is the implementation, not yet started.
- Tasks 4a, 5-9: not started. Task 5 is hard-blocked on Task 4a landing.