# 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.

**Follow-up note (live operator testing, post-deploy) — CORRECTED, see Task 10.** An initial pass at this (below) wrongly concluded FADE's variable multi-second delay was just an uncapped `transition_duration_ms` value. The user confirmed the duration param was never touched (stayed at its default, 500) across repeated tests, while the time for the switch to actually reflect on the live output varied run to run: ~7-8s, then 10s, then 20s, then 13s. That rules out the "operator entered a big number" explanation outright — this is a real, variable latency bug specific to FADE, not a config gap. Reading `dispatcher.py` turned up a directly relevant precedent already in the code's own comments (`build_composition_pipeline`, the `raw_queue` element, ~line 752-776): this exact symptom shape — "scene-switch latency that kept growing across a session (3s, then 5s, then 7s, ...)", observed unbounded up to 40s — was already root-caused once before to a pipeline-wide backlog with no drop point anywhere from the compositor to MediaMTX, and partially fixed via a leaky, time-bounded (500ms) queue between the compositor and the encoder. That fix bounds the *aggregate post-compositor* backlog, but doesn't touch the *per-source, pre-compositor* path — and FADE's mechanism (`animate_pad_property`, `dispatcher.py:915`) schedules its `GstController` interpolation using absolute pipeline running-time (`pipeline.get_clock().get_time() - pipeline.get_base_time()` at dispatch), not wall-clock time. If a given source's own buffers are running behind the pipeline's live edge (RTSP jitter/network jitter, decode backlog upstream of the compositor — not covered by the `raw_queue` fix), the control binding doesn't visibly start ramping until that source's own stream-time catches up to the scheduled start timestamp — which varies with whatever backlog existed on that specific source at the moment of the switch. CUT's `_pin_final_value` sets the property directly, unconditionally, so it's not exposed to this at all — consistent with CUT staying flat at ~0.5-1s while FADE varies. **This is a real hypothesis, not a confirmed root cause** — logged as Task 10 below, not resolved here, per the user's explicit call to dig further before treating it as understood.

**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.

**Checkpoint C resolved**: dedicated `pipeline_heartbeat_at` field (not a reuse of `Composition.updated_at`), written at most once per 5 seconds regardless of tick rate (time-based cadence, not every-tick or every-Nth-tick).

**Files:**
- `src/apps/compositor/models.py` — 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 `pipeline_heartbeat_at` at most once per 5 seconds (track the last-written wall-clock time in a local variable across loop iterations, skip the write if under 5s since the last one — a single `UPDATE ... WHERE id=...` per write, no read-then-write).
- 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 — RESOLVED**: dedicated `pipeline_heartbeat_at` field, time-based
> write cadence (max once per 5s). The spec left this open deliberately (Open
> Question 5); decided during Build rather than pre-supposed in the original Plan.

---

### 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-compositor-worker.yml`'s pod spec (the Task 4a rename of `sink-worker.yml`). 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-compositor-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-compositor-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: 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-compositor-worker.yml` — CPU limit currently 6 cores against a 6-slot `--concurrency` (raised from 4 by Task 4a); review whether the limit should rise to match, or whether the headroom slots' own short-task nature means the existing limit already covers real worst-case usage (worst-case CPU is bounded by concurrent `compositor.run` count, not raw concurrency slots — see Task 4a's own reasoning for why the limit wasn't scaled 1:1).
- `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.
- `plevion-k8s/services/sink/base/workers/sink-recording-worker.yml` — new in Task 4a, its resource sizing is an unvalidated starting guess; review once real concurrent-recording load exists.

**Acceptance:**
- [ ] A documented sizing decision (in this plan or a short follow-up note) for all three workers' CPU limits, grounded in real observed per-composition/per-overlay/ per-recording 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."

---

### Task 10 — Investigate FADE-transition variable onset delay

**Added during Build**, found via live operator testing after Task 3's own measurement was already written up (see Task 3's corrected follow-up note above — an initial "uncapped duration input" theory was wrong, ruled out by the operator confirming `transition_duration_ms` was never touched). Real symptom: CUT switches consistently land in ~0.5-1s (matches Task 3's measured dispatch latency); FADE switches, same composition, same unchanged 500ms duration, visibly took ~7-8s, then 10s, then 20s, then 13s across repeated attempts.

**Not scheduled for Build yet — investigation only, sequenced after Tasks 1-9** (this plan's own already-approved work takes priority; this is a newly-found, not-yet- understood issue, not a blocker for anything already in flight).

**Working hypothesis** (see Task 3's follow-up note for the full reasoning, not repeated here): FADE's `animate_pad_property` (`dispatcher.py:915`) schedules its `GstController` interpolation against absolute pipeline running-time computed at dispatch (`pipeline.get_clock().get_time() - pipeline.get_base_time()`), not wall-clock time — so if a specific source's own buffers are running behind the pipeline's live edge (pre-compositor decode/jitter backlog, upstream of the `raw_queue` fix that already bounds the *aggregate post-compositor* path — see that queue's own comments in `dispatcher.py` for the prior, structurally similar "scene-switch latency kept growing across a session" incident), the animation's visible onset is delayed until that source's own stream-time catches up. CUT's `_pin_final_value` sets the property directly and isn't exposed to this. **Unconfirmed** — needs real investigation, not assumed correct.

**Files (investigation, not yet a fix):**
- `sink/src/apps/compositor/services/dispatcher.py` — `animate_pad_property`, `_toggle_pad`, the `raw_queue`/backlog-handling comments (~line 752-781) for the prior related incident's full context.
- Live: `gst-launch`/pipeline latency query (`GST_DEBUG=GST_TRACER:7` or a `latency` tracer, or simplest: log each source's own jitterbuffer/queue current level at switch time) during a reproduction attempt, to confirm or rule out per-source backlog as the actual cause before designing a fix.

**Acceptance (investigation phase):**
- [ ] Reproduced live with instrumentation showing the actual state (buffer level/ running-time skew) of the source(s) involved in a slow FADE vs. a fast one, not just timestamps in application logs.
- [ ] A written conclusion: confirms or rules out the running-time-skew hypothesis above, and if confirmed, a follow-up fix task is scoped (e.g. sync `animate_pad_property`'s start timestamp to the specific pad's own current stream-time instead of the pipeline's, or extend the `raw_queue`-style leaky-bound fix to the pre-compositor path).

**Verification:** Manual/live only, per the same real-composition-testing bar as any other `dispatcher.py`-adjacent work in this plan.

---

## 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.
- **A `max_value` cap on `transition_duration_ms`** (frontend input + backend serializer) — still a legitimate independent gap (no upper bound exists anywhere in the path), but no longer believed to explain the FADE latency symptom itself; see Task 10.

## 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-compositor-worker.yml`, `sink-recording-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), `MVP-26` for Task 4a (Checkpoint B's own resolution — the 3-queue split implementation, also added during Build), and `MVP-27` for Task 10 (FADE onset-delay investigation, added during Build after MVP-19's own follow-up testing).

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). Its initial FADE-duration follow-up note was corrected after further live testing — see Task 3's follow-up note and Task 10 (`MVP-27`) for the real, still-unconfirmed finding.
- **Checkpoint B resolved**: 3-queue split (per-domain), see Task 4's own updated entry above. **MVP-26/Task 4a done** — sink PR [#130](https://github.com/plevion-dev/sink/pull/130) and plevion-k8s PR [#46](https://github.com/plevion-dev/plevion-k8s/pull/46) merged, image built (`sink`/`sink-capturer` `v0.0.227-9b1202a`), `plevion-k8s` bumped (`ea8c691`), **deployed and verified live** — `sink-worker` retired, `sink-compositor-worker-local`/`sink-recording-worker-local` confirmed running and healthy, a leftover un-pruned old `sink-worker-local` Deployment found and manually deleted post-redeploy (`kubectl apply -k` doesn't prune resources dropped from a kustomization — worth a follow-up if this bites again). Live operator testing post-deploy (a real composition, multiple scene switches including transitions) confirmed healthy — informal, not a substitute for Task 5's own real recovery-scenario test, but a good sign.
- **MVP-27** (Task 10, FADE onset-delay investigation) — created, not started; sequenced after Tasks 1-9, not blocking anything else.
- **Checkpoint C resolved**: dedicated `pipeline_heartbeat_at` field, time-based write cadence (max once per 5s) — see Task 5's own updated entry above. Task 5 is now fully unblocked and ready for Build.
- Tasks 5-9: not started.