Stage managed broadcast before starting ingest (#11)

## Summary

- Reconcile, create if needed, and bind exactly one marked unlisted
broadcast before FFmpeg starts ingest.
- Fail closed before ingest when an unmarked recoverable broadcast is
already bound to the reusable stream.
- Reuse the prepared lifecycle state after ingest becomes active,
avoiding a redundant reconciliation pass.
- Include API operation, HTTP status, and structured reason values in
redacted supervisor failures.
- Update the executive flow, Pi runbook, architecture diagram, lifecycle
design, and changelog.

## Root Cause

The deployed controller started FFmpeg before it could create the first
marked replacement event. A legacy unmarked event with auto-start and
auto-stop enabled therefore became live as soon as ingest appeared.
`liveBroadcasts.insert` then returned `User requests exceed the rate
limit.`, the pre-public fail-closed path stopped FFmpeg, and YouTube
ended the legacy event seven seconds after it started.

The durable write-ahead generation remained safe and no new marked
broadcast ID was recorded, but the quota backoff repeated the same
visible start/stop sequence.

## Recovery Design

- Source probe remains the first gate; a camera that is already
unavailable creates nothing.
- Non-public recovery stages one marked unlisted event and durably
verifies binding before FFmpeg starts.
- Active and upcoming inventory blocks any unmarked event bound to the
reusable stream.
- Lost insert responses reconcile by the existing generation marker
before another insert.
- Insert throttling preserves the same pending generation.
- Bind failure preserves the returned broadcast ID and retries binding
without reinserting.
- Active/good ingest and fresh FFmpeg media remain mandatory for
`testing`, `live`, and public promotion.
- A previously verified public event still starts media without making
control-plane availability a hard pre-ingest dependency.

## Rollout And Rollback

Rollout will stop the Pi service, create a private timestamped snapshot,
deploy the merged scripts, clear only the expired retry deadline, and
restart the service. If insert remains rate-limited, FFmpeg must not
start and the same pending generation must remain under bounded
cooldown. If insert succeeds, exactly one marked unlisted event must be
bound before ingest, then transition to live and public after health
gates.

Local rollback is valid only before a new remote event is staged. Once
staging succeeds, the prior version must remain stopped because it
cannot understand the new marked generation; no automatic YouTube
deletion, completion, or unbinding is part of rollback.

## Verification

- `pytest -q`: 102 passed
- `ruff check .`: passed
- `python -m py_compile` for all runtime scripts: passed
- Markdownlint CLI 0.18.1 across repository Markdown: passed
- `git diff --check`: passed
- Commit is GPG-signed
- Two distinct-family architecture reviews completed; the first changed
sequencing to pre-ingest staging, and the revised review surfaced no
load-bearing blocker
- Distinct-family code review was attempted twice but returned no
content; hosted CI and PR review remain required before merge

Author: sumitake

CHANGELOG.md

@@ -2,6 +2,14 @@
 
 All notable project changes are recorded here. The README shows only the most recent entry.
 
+## 2026-07-11 - Pre-Ingest Broadcast Staging
+
+- Reordered every recovery, including cached public state, so one marked broadcast is reconciled, created unlisted if needed, and bound before FFmpeg starts ingest; startup API failures now fail closed.
+- Added fail-closed detection for unmarked recoverable broadcasts bound to the reusable stream, preventing unintended legacy auto-start events.
+- Reused the prepared lifecycle state after ingest becomes active, while revalidating cached public events and reconciling a replacement when YouTube confirms the cached event is terminal, missing, or rebound.
+- Added redacted API operation, HTTP status, and structured reason details to supervisor failures for actionable quota diagnostics.
+- Added regression coverage for offline staging, insert and bind partial failures, stable generation reuse, public-stream fallback, and CLI gating.
+
 ## 2026-07-10 - Idempotent YouTube Lifecycle Recovery
 
 - Added durable instance and generation markers so one nonterminal broadcast is reconciled and reused across camera, FFmpeg, service, and host restarts.

README.md

@@ -12,7 +12,7 @@ The project runs FFmpeg under systemd, optionally reuses OBS profile data for ca
 
 YouTube AutoEncoder turns a dedicated Linux device into an unattended YouTube streaming encoder. The core service performs three jobs:
 
-- Validate the camera and FFmpeg media path before creating or transitioning a YouTube broadcast.
+- Validate the camera source, stage one marked unlisted broadcast, and require healthy media before lifecycle transitions.
 - Push the stream with FFmpeg using low-CPU video copy mode or explicit transcode mode.
 - Reconcile one marked YouTube broadcast, stage it unlisted, and publish it only after live health is confirmed.
 
@@ -78,18 +78,18 @@ OBS compatibility mode can also read and update:
 2. The supervisor acquires its single-instance lock and honors any persisted recovery deadline.
 3. The source and ingest URLs are resolved from direct settings or OBS compatibility files.
 4. FFprobe checks the RTSP source; unavailable sources use bounded source backoff without a broadcast insert.
-5. FFmpeg starts, emits machine-readable progress, and pushes to the reusable YouTube stream.
-6. One-shot API checks wait for both recent media progress and active YouTube ingest.
-7. The helper reconciles local state with exact instance and generation markers, then reuses one nonterminal broadcast or creates one unlisted event when no recoverable event exists.
-8. Before and after each `testing` or `live` transition, the supervisor rechecks FFmpeg progress and YouTube ingest.
-9. Two consecutive healthy `live` observations are required before visibility changes to `YTA_YOUTUBE_LIVE_PRIVACY`.
-10. The helper verifies the privacy readback, clears recovery state, and the supervisor stops nonessential API polling.
-11. FFmpeg remains supervised until source loss, process failure, service stop, rotation, or host interruption.
-12. Failures persist a class-specific cooldown and restart the same broadcast without completing it.
+5. Before ingest starts, the helper lists recoverable events, blocks any unmarked event bound to the reusable stream, and reuses or stages exactly one marked unlisted broadcast.
+6. The helper durably stores the broadcast ID and verifies binding before the supervisor starts FFmpeg.
+7. FFmpeg emits machine-readable progress and pushes to the reusable YouTube stream.
+8. One-shot API checks wait for both recent media progress and active YouTube ingest.
+9. Before and after each `testing` or `live` transition, the supervisor rechecks FFmpeg progress and YouTube ingest.
+10. Two consecutive healthy `live` observations are required before visibility changes to `YTA_YOUTUBE_LIVE_PRIVACY`.
+11. The helper verifies the privacy readback, clears recovery state, and the supervisor stops nonessential API polling.
+12. FFmpeg remains supervised until source loss, process failure, service stop, rotation, or host interruption; failures persist a class-specific cooldown and reuse the same broadcast without completing it.
 
 ### Recovery Behavior
 
-Every recovery path first preserves ownership and retry state. Media must be fresh and YouTube ingest active before reconciliation can create or transition anything.
+Every recovery path first preserves ownership and retry state. After a passing source probe, the YouTube control plane must reconcile ownership and binding before FFmpeg starts, including for cached public state. Fresh media and active YouTube ingest remain mandatory before `testing`, `live`, or public promotion.
 
 See the [recovery state machine](docs/architecture-and-flows.md#recovery-state-machine) for startup, managed-generation, and durable-cooldown transitions.
 
@@ -99,11 +99,13 @@ See the [recovery state machine](docs/architecture-and-flows.md#recovery-state-m
 | Camera loses power during stream | FFmpeg exits; the same broadcast and watch URL remain; source backoff continues until the camera returns. |
 | Host reboots | systemd restarts the service, the durable cache is reconciled, and the same nonterminal event resumes. |
 | FFmpeg exits or stalls | The child and any in-flight API helper stop; the same event is retained for retry. |
-| YouTube ingest does not become active | No insert or transition occurs; the attempt uses source/encoder backoff. |
-| YouTube API rate limit or outage | The retry class and deadline persist. A previously verified public stream can continue without control-plane mutation. |
+| YouTube ingest does not become active | The single staged unlisted event is retained; no transition or publication occurs. |
+| YouTube API rate limit or outage before FFmpeg starts | Startup fails closed, no ingest begins, and the retry class and deadline persist. |
+| YouTube API rate limit or outage after validated public ingest is active | The already-running public stream can continue without control-plane mutation. |
+| Unmarked event is bound to the reusable stream | Reconciliation fails closed before FFmpeg starts, preventing an unintended legacy auto-start. |
 | Ambiguous or unknown remote state | Reconciliation fails closed and creates nothing until the ambiguity is resolved. |
 | OAuth access token expires | API helper refreshes from the stored refresh token. |
-| Previous broadcast is `complete`, `revoked`, or confirmed missing | One new generation may be created after ingest is active. |
+| Previous broadcast is `complete`, `revoked`, or confirmed missing | One new unlisted generation may be staged after the source probe passes. |
 
 Recovery deadlines survive service and host restarts. Exponential backoff uses these class floors and caps:
 
@@ -408,6 +410,7 @@ youtube-autoencoder-api complete
 - YouTube Live must already be enabled on the channel. New or restricted channels may not be allowed to stream immediately.
 - The YouTube Data API flow requires OAuth user consent. A simple API key is not enough for creating, binding, or transitioning live broadcasts.
 - Google OAuth app restrictions can block authorization if the app is limited to an organization that does not include the streaming account.
+- A source that passes its probe and then fails can leave one marked unlisted upcoming event. Recovery reuses that event; it does not create another generation.
 - YouTube API quota, API outages, or account policy restrictions can prevent lifecycle operations even when FFmpeg is healthy.
 - Recovery reuses one exactly marked nonterminal broadcast. A new generation is permitted only after the previous managed event is `complete`, `revoked`, or confirmed missing.
 - Broadcasts created by older releases do not contain ownership markers. They are not adopted or deleted automatically; inventory and clean up legacy duplicates separately after verifying their lifecycle and watch URLs.
@@ -425,9 +428,9 @@ youtube-autoencoder-api complete
 
 Only the most recent changelog entry is shown here. See `CHANGELOG.md` for full history.
 
-### 2026-07-10 - Idempotent YouTube Lifecycle Recovery
+### 2026-07-11 - Pre-Ingest Broadcast Staging
 
-- Added exact broadcast reconciliation, FFmpeg progress supervision, persisted backoff, explicit-only completion, and verified unlisted-to-public promotion while preserving one watch URL across failures.
+- Reconciles every startup before FFmpeg ingest, stages and binds one marked unlisted event when needed, blocks unmarked bound conflicts, revalidates cached public events after ingest, and logs structured API operation and rate-limit details without exposing credentials.
 
 ## Repository Layout
 

bin/youtube-autoencoder

@@ -73,21 +73,32 @@ class ApiCommandError(RuntimeError):
         self,
         *,
         returncode: int,
+        operation: str | None = None,
         payload: dict[str, Any] | None = None,
         output: str = "",
     ) -> None:
         self.returncode = returncode
+        self.operation = operation
         self.payload = payload or {}
         self.http_status = self.payload.get("http_status")
         self.reasons = tuple(str(reason) for reason in self.payload.get("reasons") or [])
         self.retry_class = str(self.payload.get("retry_class") or "fatal")
+        self.public_fallback_allowed = self.payload.get("public_fallback_allowed") is not False
         retry_after = self.payload.get("retry_after")
         try:
             self.retry_after = float(retry_after) if retry_after is not None else None
         except (TypeError, ValueError):
             self.retry_after = None
         message = str(self.payload.get("message") or output or "YouTube API helper failed")
-        super().__init__(redact_text(message))
+        details = []
+        if operation:
+            details.append(f"operation={operation}")
+        if self.http_status is not None:
+            details.append(f"http_status={self.http_status}")
+        if self.reasons:
+            details.append(f"reasons={','.join(self.reasons)}")
+        suffix = f" ({' '.join(details)})" if details else ""
+        super().__init__(redact_text(message + suffix))
 
 
 @dataclasses.dataclass
@@ -401,11 +412,17 @@ def json_object_from_output(output: str) -> dict[str, Any] | None:
     return None
 
 
-def parse_api_result(returncode: int, stdout: str, stderr: str) -> dict[str, Any]:
+def parse_api_result(
+    returncode: int,
+    stdout: str,
+    stderr: str,
+    *,
+    operation: str | None = None,
+) -> dict[str, Any]:
     if returncode != 0:
         payload = json_object_from_output(stderr) or json_object_from_output(stdout)
         output = (stdout + "\n" + stderr).strip()
-        raise ApiCommandError(returncode=returncode, payload=payload, output=output)
+        raise ApiCommandError(returncode=returncode, operation=operation, payload=payload, output=output)
     output = stdout.strip()
     if not output:
         return {}
@@ -414,23 +431,27 @@ def parse_api_result(returncode: int, stdout: str, stderr: str) -> dict[str, Any
     except json.JSONDecodeError as exc:
         raise ApiCommandError(
             returncode=returncode,
+            operation=operation,
             payload={"message": f"YouTube API helper returned invalid JSON: {exc}", "retry_class": "fatal"},
         ) from exc
     if not isinstance(value, dict):
         raise ApiCommandError(
             returncode=returncode,
+            operation=operation,
             payload={"message": "YouTube API helper did not return a JSON object", "retry_class": "fatal"},
         )
     return value
 
 
 def api_command(args: list[str], *, timeout: int = 90) -> dict[str, Any]:
     cmd = [env("YTA_YOUTUBE_API", "youtube-autoencoder-api"), *args]
+    operation = args[0] if args else None
     try:
         result = subprocess.run(cmd, text=True, capture_output=True, timeout=timeout, check=False)
     except subprocess.TimeoutExpired as exc:
         raise ApiCommandError(
             returncode=-1,
+            operation=operation,
             payload={
                 "message": f"YouTube API command timed out after {timeout}s",
                 "retry_class": "api",
@@ -439,9 +460,10 @@ def api_command(args: list[str], *, timeout: int = 90) -> dict[str, Any]:
     except OSError as exc:
         raise ApiCommandError(
             returncode=-1,
+            operation=operation,
             payload={"message": f"could not start YouTube API helper: {exc}", "retry_class": "fatal"},
         ) from exc
-    return parse_api_result(result.returncode, result.stdout, result.stderr)
+    return parse_api_result(result.returncode, result.stdout, result.stderr, operation=operation)
 
 
 def ffmpeg_args(
@@ -582,6 +604,7 @@ def api_command_while_streaming(
     timeout: int,
 ) -> dict[str, Any]:
     cmd = [env("YTA_YOUTUBE_API", "youtube-autoencoder-api"), *args]
+    operation = args[0] if args else None
     with (
         tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as stdout_file,
         tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as stderr_file,
@@ -596,6 +619,7 @@ def api_command_while_streaming(
         except OSError as exc:
             raise ApiCommandError(
                 returncode=-1,
+                operation=operation,
                 payload={"message": f"could not start YouTube API helper: {exc}", "retry_class": "fatal"},
             ) from exc
         deadline = time.monotonic() + timeout
@@ -607,6 +631,7 @@ def api_command_while_streaming(
                 if time.monotonic() >= deadline:
                     raise ApiCommandError(
                         returncode=-1,
+                        operation=operation,
                         payload={
                             "message": f"YouTube API command timed out after {timeout}s",
                             "retry_class": "api",
@@ -622,7 +647,7 @@ def api_command_while_streaming(
         except BaseException:
             stop_process(api_process)
             raise
-    return parse_api_result(api_process.returncode, stdout, stderr)
+    return parse_api_result(api_process.returncode, stdout, stderr, operation=operation)
 
 
 def poll_interval() -> float:
@@ -662,6 +687,17 @@ def current_observation(runtime: StreamRuntime, state: dict[str, Any]) -> dict[s
     ensure_encoder_running(runtime)
     stream = api_command_while_streaming(["stream-status", stream_id], runtime, timeout=45)
     broadcast = api_command_while_streaming(["broadcast-status", broadcast_id], runtime, timeout=45)
+    bound_stream_id = str(broadcast.get("bound_stream_id") or "")
+    if bound_stream_id != stream_id:
+        raise ApiCommandError(
+            returncode=78,
+            operation="broadcast-status",
+            payload={
+                "message": f"managed broadcast is bound to a different stream: {bound_stream_id or '<empty>'}",
+                "retry_class": "ambiguous",
+                "public_fallback_allowed": False,
+            },
+        )
     ensure_encoder_running(runtime)
     return {
         "stream_status": stream.get("stream_status"),
@@ -686,7 +722,7 @@ def wait_for_active_stream(runtime: StreamRuntime) -> dict[str, Any]:
         wait_with_runtime(runtime, poll_interval())
 
 
-def reconcile_command_args() -> list[str]:
+def reconcile_command_args(*, offline_create: bool = False) -> list[str]:
     args = [
         "reconcile-broadcast",
         "--allow-create",
@@ -698,6 +734,8 @@ def reconcile_command_args() -> list[str]:
     title = os.environ.get("YTA_YOUTUBE_TITLE")
     if title:
         args.extend(["--title", title])
+    if offline_create:
+        args.append("--offline-create")
     return args
 
 
@@ -789,9 +827,17 @@ def publish_when_healthy(runtime: StreamRuntime, initial_state: dict[str, Any])
     raise EncoderStopped("YouTube stream did not remain healthy long enough for publication")
 
 
-def manage_youtube_lifecycle(runtime: StreamRuntime) -> dict[str, Any]:
+def manage_youtube_lifecycle(
+    runtime: StreamRuntime,
+    prepared_state: dict[str, Any] | None = None,
+) -> dict[str, Any]:
     active_stream = wait_for_active_stream(runtime)
-    state = api_command_while_streaming(reconcile_command_args(), runtime, timeout=120)
+    if prepared_state is None:
+        state = api_command_while_streaming(reconcile_command_args(), runtime, timeout=120)
+    else:
+        state = dict(prepared_state)
+        if verified_public_state(state):
+            state = revalidate_public_state(runtime, state)
     stream_id = str(state.get("stream_id") or state.get("last_stream_id") or "")
     broadcast_id = str(state.get("broadcast_id") or state.get("last_broadcast_id") or "")
     if not stream_id or not broadcast_id:
@@ -820,11 +866,50 @@ def verified_public_state(state: dict[str, Any]) -> bool:
     return lifecycle == "live" and privacy == "public" and bool(stream_id and broadcast_id)
 
 
+def disable_public_fallback(error: ApiCommandError) -> ApiCommandError:
+    payload = dict(error.payload)
+    payload["public_fallback_allowed"] = False
+    return ApiCommandError(
+        returncode=error.returncode,
+        operation=error.operation,
+        payload=payload,
+    )
+
+
+def reconcile_invalid_public_state(runtime: StreamRuntime) -> dict[str, Any]:
+    try:
+        return api_command_while_streaming(reconcile_command_args(), runtime, timeout=120)
+    except ApiCommandError as error:
+        raise disable_public_fallback(error) from error
+
+
+def revalidate_public_state(runtime: StreamRuntime, state: dict[str, Any]) -> dict[str, Any]:
+    try:
+        observation = current_observation(runtime, state)
+    except ApiCommandError as error:
+        if error.public_fallback_allowed:
+            raise
+        log("cached public broadcast is no longer recoverable; reconciling replacement")
+        return reconcile_invalid_public_state(runtime)
+
+    if observation.get("lifecycle") == "live" and observation.get("privacy") == "public":
+        refreshed = dict(state)
+        refreshed["lifecycle"] = "live"
+        refreshed["privacy"] = "public"
+        return refreshed
+
+    log(
+        "cached public broadcast is no longer live/public; reconciling replacement "
+        f"lifecycle={observation.get('lifecycle')} privacy={observation.get('privacy')}"
+    )
+    return reconcile_invalid_public_state(runtime)
+
+
 def manage_with_public_fallback(runtime: StreamRuntime, cached_state: dict[str, Any]) -> dict[str, Any]:
     try:
-        return manage_youtube_lifecycle(runtime)
+        return manage_youtube_lifecycle(runtime, prepared_state=cached_state)
     except ApiCommandError as error:
-        if not verified_public_state(cached_state):
+        if not verified_public_state(cached_state) or not error.public_fallback_allowed:
             raise
         ensure_encoder_running(runtime)
         retry_class = classify_api_error(error)
@@ -912,6 +997,12 @@ def read_recovery_state() -> dict[str, Any]:
     return api_command(["state"], timeout=30)
 
 
+def prepare_youtube_lifecycle(cached_state: dict[str, Any]) -> dict[str, Any]:
+    if not lifecycle_enabled():
+        return dict(cached_state)
+    return api_command(reconcile_command_args(offline_create=True), timeout=120)
+
+
 def supervise_stream(runtime: StreamRuntime, *, started_at: float, max_runtime: float) -> int:
     while not stopping:
         drain_ffmpeg_output(runtime, timeout=1.0)
@@ -939,6 +1030,8 @@ def run_once(cached_state: dict[str, Any] | None = None) -> int:
     if not source_available(input_url):
         raise SourceUnavailable("camera source probe failed")
 
+    prepared_state = prepare_youtube_lifecycle(cached_state or {})
+
     log(f"starting ffmpeg mode={env('YTA_MODE', 'copy')}")
     started = time.monotonic()
     child = subprocess.Popen(
@@ -963,7 +1056,7 @@ def run_once(cached_state: dict[str, Any] | None = None) -> int:
         if bool_env("YTA_YOUTUBE_COMPLETE_ON_EXIT", False):
             log("YTA_YOUTUBE_COMPLETE_ON_EXIT is deprecated and ignored; use the explicit complete command")
         if lifecycle_enabled():
-            manage_with_public_fallback(runtime, cached_state or {})
+            manage_with_public_fallback(runtime, prepared_state)
         return supervise_stream(runtime, started_at=started, max_runtime=max_runtime)
     finally:
         if child and child.poll() is None: