feat: bring livepeer runner Kafka events to parity with cloud-relay

[emranemran]

Summary

Follow-up to #956. That PR added websocket_connected / websocket_disconnected from the livepeer fal wrapper, but the rest of the session lifecycle (pipeline_loaded, session_created, stream_started, stream_heartbeat, stream_stopped, playback_ready, error variants) was either not firing or firing with null identifiers in livepeer mode. Cloud-relay mode (fal_app.py path) already had these events working; this PR brings livepeer mode to the same shape.

Root causes (all in the runner, `src/scope/cloud/livepeer_app.py`)

1. `FrameProcessor(...)` was constructed without `user_id` / `connection_id` / `session_id` / `connection_info`, so every event it emits (`stream_started`, `stream_heartbeat`, `stream_stopped`, etc.) carried nulls.
2. `session.manifest_id` was never populated from `job_info.manifest_id`, so there was no identifier available that matched the wrapper's `websocket_connected.connection_id`.
3. The pipeline/load body injection used the runner's random internal UUID as `connection_id` instead of the manifest_id, so `pipeline_loaded` didn't correlate.
4. `session_created` / `session_closed` are only emitted from `webrtc.py`, which isn't on the livepeer code path — they have to be emitted explicitly.
5. `connection_info` env vars (`NOMAD_DC`, `FAL_JOB_ID`, `FAL_LOG_LABELS`, etc.) weren't in the runner subprocess allowlist, so the runner couldn't rebuild the dict even if asked.

Changes

`src/scope/cloud/livepeer_app.py`

• Add `manifest_id` / `session_id` / `connection_info` fields to `LivepeerSession`; populate them right after parsing `ScopeJobInfo`.
• New `_build_connection_info()` helper that mirrors the shape built by `livepeer_fal_app.py` from env vars.
• Pass `session_id`, `user_id`, `connection_id=manifest_id`, `connection_info` into `FrameProcessor` so every downstream event from `frame_processor.py` / `pipeline_processor.py` carries identifiers that join with `websocket_connected`.
• Emit `session_created` right after `FrameProcessor.start()`, and `session_closed` right after `FrameProcessor.stop()`, using the same shape as `webrtc.py:731` and `:1150`.
• Flip the pipeline/load body injection from `session.connection_id` to `session.manifest_id`, and pass `connection_info` too.

`src/scope/cloud/livepeer_fal_app.py`

• Add `NOMAD_DC`, `FAL_JOB_ID`, `FAL_RUNNER_ID`, `FAL_LOG_LABELS`, `FAL_MACHINE_TYPE` to `env_allowlist` so the runner subprocess can reconstruct `connection_info`.

Test plan

• `uv run ruff check src/` and `ruff format --check src/` pass (already verified)
• `uv run daydream-scope` starts without import/init errors
• `./test-cloud-connect.sh --skip-push` after CI build-cloud succeeds → exit 0 with CONNECTED
• Manual UI test: connect to cloud, load pipeline (`longlive`), start stream ~30s, disconnect
• ClickHouse query on `scope_cloud_events` filtered by `user_id` and `connection_id = <manifest_id>` shows all of: `websocket_connected`, `pipeline_load_start`, `pipeline_loaded`, `session_created`, `stream_started`, ≥2 `stream_heartbeat`, `stream_stopped`, `session_closed`, `websocket_disconnected` — all sharing the same connection_id
• Regression check on cloud-relay path (`fal_app.py` deploy): no changes, events unchanged

Not in scope

• Local-scope-side events (`session_created` from WebRTC path on user's laptop, `session_closed`, webrtc connection errors) still require the user's local env to have `KAFKA_*` configured; that's a config concern, not code.
• `/api/v1/session/start` not being livepeer-compatible (`mcp_router.py:252` TODO) — untouched; only affects headless test flows.

🤖 Co-authored with Claude Code

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 090198ca-69ef-4e23-81bb-1d9b8a7d152f

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/livepeer-runner-kafka-parity

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[github-actions]

🚀 fal.ai Preview Deployment

App ID	daydream/scope-pr-969--preview
WebSocket	wss://fal.run/daydream/scope-pr-969--preview/ws
Commit	291c5c9

Livepeer Runner

App ID	daydream/scope-livepeer-pr-969--preview
WebSocket	wss://fal.run/daydream/scope-livepeer-pr-969--preview/ws
Auth	private

Testing Livepeer Mode

SCOPE_CLOUD_MODE=livepeer SCOPE_CLOUD_APP_ID="daydream/scope-livepeer-pr-969--preview/ws" uv run daydream-scope

[j0sh]

Pretty sure manifest_id is the same as the connection_id above

[j0sh]

... oh dear it's not. Well that's a simple change then

[emranemran]

i made manifest id to be same as connection id and the old conenction id should be gone now.

[j0sh]

I guess the manifest_id does not need to be passed in anymore if that's the case? Would simplify some of the ad hoc checks around here.

[emranemran]

Yeah that's true. I'll clean this up in a follow up and file a ticket so i don't forget.