encoder: revalidate cached public broadcasts

Author: sumitake

CHANGELOG.md

@@ -6,7 +6,7 @@ All notable project changes are recorded here. The README shows only the most re
 
 - Reordered recovery so one marked unlisted broadcast is reconciled, created if needed, and bound before FFmpeg starts ingest.
 - 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 to avoid a redundant reconciliation pass.
+- 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.
 

README.md

@@ -429,7 +429,7 @@ Only the most recent changelog entry is shown here. See `CHANGELOG.md` for full
 
 ### 2026-07-11 - Pre-Ingest Broadcast Staging
 
-- Stages and binds one marked unlisted event before FFmpeg ingest, blocks unmarked bound conflicts, and logs structured API operation and rate-limit details without exposing credentials.
+- Stages and binds one marked unlisted event before FFmpeg ingest, 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

@@ -83,6 +83,7 @@ class ApiCommandError(RuntimeError):
         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
@@ -694,6 +695,7 @@ def current_observation(runtime: StreamRuntime, state: dict[str, Any]) -> dict[s
             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)
@@ -830,11 +832,12 @@ def manage_youtube_lifecycle(
     prepared_state: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     active_stream = wait_for_active_stream(runtime)
-    state = (
-        dict(prepared_state)
-        if prepared_state is not None
-        else 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:
@@ -863,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, 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)

bin/youtube-autoencoder-api

@@ -86,6 +86,10 @@ class ReconciliationError(RuntimeError):
     retry_class = "ambiguous"
 
 
+class RemoteBroadcastStateError(ReconciliationError):
+    public_fallback_allowed = False
+
+
 def log(message: str) -> None:
     print(f"{time.strftime('%Y-%m-%dT%H:%M:%S%z')} {message}", file=sys.stderr, flush=True)
 
@@ -185,14 +189,17 @@ def error_payload(exc: BaseException) -> dict[str, Any]:
             "retry_class": exc.retry_class,
         }
     retry_class = str(getattr(exc, "retry_class", "fatal"))
-    return {
+    payload = {
         "error": True,
         "http_status": None,
         "message": str(exc),
         "reasons": [],
         "retry_after": None,
         "retry_class": retry_class,
     }
+    if getattr(exc, "public_fallback_allowed", True) is False:
+        payload["public_fallback_allowed"] = False
+    return payload
 
 
 def client_config() -> dict[str, str]:
@@ -845,7 +852,7 @@ def broadcast_status_payload(broadcast: dict[str, Any]) -> dict[str, Any]:
 def broadcast_status(broadcast_id: str) -> dict[str, Any]:
     broadcast = broadcast_by_id(broadcast_id)
     if broadcast is None:
-        raise ReconciliationError(f"liveBroadcast not found: {broadcast_id}")
+        raise RemoteBroadcastStateError(f"liveBroadcast not found: {broadcast_id}")
     return broadcast_status_payload(broadcast)
 
 

tests/test_api.py

@@ -804,6 +804,18 @@ def test_set_privacy_command_reuses_verified_readback(load_script, monkeypatch,
     assert output["privacy"] == "public"
 
 
+def test_missing_broadcast_status_disables_cached_public_fallback(load_script, monkeypatch):
+    api = load_script("youtube-autoencoder-api", "yta_api_missing_broadcast_status")
+    monkeypatch.setattr(api, "broadcast_by_id", lambda _broadcast_id: None)
+
+    with pytest.raises(api.ReconciliationError, match="not found") as raised:
+        api.broadcast_status("missing-broadcast")
+
+    payload = api.error_payload(raised.value)
+    assert payload["retry_class"] == "ambiguous"
+    assert payload["public_fallback_allowed"] is False
+
+
 def test_retry_state_updates_preserve_broadcast_identity(load_script, monkeypatch, tmp_path):
     api = load_script("youtube-autoencoder-api", "yta_api_retry_state")
     configure_reconciliation(api, monkeypatch, tmp_path)

tests/test_supervisor.py

@@ -467,6 +467,7 @@ def test_current_observation_rejects_broadcast_rebound_to_different_stream(load_
         )
 
     assert raised.value.retry_class == "ambiguous"
+    assert raised.value.public_fallback_allowed is False
 
 
 @pytest.mark.parametrize(
@@ -725,6 +726,134 @@ def test_prepared_lifecycle_reuses_staged_state_after_ingest(load_script, monkey
     assert state["privacy"] == "public"
 
 
+def test_cached_public_broadcast_is_reconciled_when_remote_event_completed(load_script, monkeypatch):
+    supervisor = load_script("youtube-autoencoder", "yta_public_completed_reconcile")
+    events = []
+    runtime = supervisor.StreamRuntime(
+        process=FakeFfmpegProcess(returncode=None),
+        selector=FakeSelector(),
+        watchdog=supervisor.ProgressWatchdog(
+            started_at=supervisor.time.monotonic(),
+            timeout=30.0,
+            last_progress_at=supervisor.time.monotonic(),
+        ),
+    )
+    cached = {
+        "stream_id": "stream-1",
+        "broadcast_id": "broadcast-old",
+        "lifecycle": "live",
+        "privacy": "public",
+    }
+    replacement = {
+        "stream_id": "stream-1",
+        "broadcast_id": "broadcast-new",
+        "lifecycle": "ready",
+        "privacy": "unlisted",
+    }
+
+    monkeypatch.setattr(
+        supervisor,
+        "wait_for_active_stream",
+        lambda _runtime: events.append("active") or {"stream_id": "stream-1"},
+    )
+    monkeypatch.setattr(
+        supervisor,
+        "current_observation",
+        lambda _runtime, _state: events.append("revalidate")
+        or {
+            "stream_status": "active",
+            "health": "good",
+            "lifecycle": "complete",
+            "privacy": "public",
+            "encoder_alive": True,
+            "media_fresh": True,
+        },
+    )
+
+    def command(args, _runtime, timeout):
+        assert args[0] == "reconcile-broadcast"
+        assert "--offline-create" not in args
+        assert timeout == 120
+        events.append("reconcile")
+        return replacement
+
+    monkeypatch.setattr(supervisor, "api_command_while_streaming", command)
+    monkeypatch.setattr(
+        supervisor,
+        "reconcile_lifecycle",
+        lambda _runtime, state: events.append("lifecycle") or {**state, "lifecycle": "live"},
+    )
+    monkeypatch.setattr(
+        supervisor,
+        "publish_when_healthy",
+        lambda _runtime, state: events.append("publish") or {**state, "privacy": "public"},
+    )
+
+    state = supervisor.manage_youtube_lifecycle(runtime, prepared_state=cached)
+
+    assert events == ["active", "revalidate", "reconcile", "lifecycle", "publish"]
+    assert state["broadcast_id"] == "broadcast-new"
+
+
+def test_cached_public_broadcast_is_revalidated_after_ingest(load_script, monkeypatch):
+    supervisor = load_script("youtube-autoencoder", "yta_public_revalidated")
+    events = []
+    runtime = supervisor.StreamRuntime(
+        process=FakeFfmpegProcess(returncode=None),
+        selector=FakeSelector(),
+        watchdog=supervisor.ProgressWatchdog(
+            started_at=supervisor.time.monotonic(),
+            timeout=30.0,
+            last_progress_at=supervisor.time.monotonic(),
+        ),
+    )
+    cached = {
+        "stream_id": "stream-1",
+        "broadcast_id": "broadcast-1",
+        "lifecycle": "live",
+        "privacy": "public",
+    }
+
+    monkeypatch.setattr(
+        supervisor,
+        "wait_for_active_stream",
+        lambda _runtime: events.append("active") or {"stream_id": "stream-1"},
+    )
+    monkeypatch.setattr(
+        supervisor,
+        "current_observation",
+        lambda _runtime, _state: events.append("revalidate")
+        or {
+            "stream_status": "active",
+            "health": "good",
+            "lifecycle": "live",
+            "privacy": "public",
+            "encoder_alive": True,
+            "media_fresh": True,
+        },
+    )
+    monkeypatch.setattr(
+        supervisor,
+        "api_command_while_streaming",
+        lambda *_args, **_kwargs: pytest.fail("valid public broadcast was reconciled again"),
+    )
+    monkeypatch.setattr(
+        supervisor,
+        "reconcile_lifecycle",
+        lambda _runtime, state: events.append("lifecycle") or state,
+    )
+    monkeypatch.setattr(
+        supervisor,
+        "publish_when_healthy",
+        lambda _runtime, state: events.append("publish") or state,
+    )
+
+    state = supervisor.manage_youtube_lifecycle(runtime, prepared_state=cached)
+
+    assert events == ["active", "revalidate", "lifecycle", "publish"]
+    assert state == cached
+
+
 def test_nonpublic_state_is_staged_before_ffmpeg_starts(load_script, monkeypatch):
     supervisor = load_script("youtube-autoencoder", "yta_preingest_stage_order")
     events = []
@@ -852,6 +981,48 @@ def capture(args, _runtime, timeout):
     assert calls[0][1] == 30
 
 
+def test_invalidated_public_stream_does_not_enter_api_fallback(load_script, monkeypatch):
+    supervisor = load_script("youtube-autoencoder", "yta_invalid_public_no_fallback")
+    runtime = supervisor.StreamRuntime(
+        process=FakeFfmpegProcess(returncode=None),
+        selector=FakeSelector(),
+        watchdog=supervisor.ProgressWatchdog(
+            started_at=supervisor.time.monotonic(),
+            timeout=30.0,
+            last_progress_at=supervisor.time.monotonic(),
+        ),
+    )
+    cached = {
+        "stream_id": "stream-1",
+        "broadcast_id": "broadcast-1",
+        "lifecycle": "live",
+        "privacy": "public",
+    }
+    failure = supervisor.ApiCommandError(
+        returncode=75,
+        payload={
+            "message": "rate limited while replacing completed broadcast",
+            "retry_class": "quota",
+            "public_fallback_allowed": False,
+        },
+    )
+    monkeypatch.setattr(
+        supervisor,
+        "manage_youtube_lifecycle",
+        lambda _runtime, prepared_state=None: (_ for _ in ()).throw(failure),
+    )
+    monkeypatch.setattr(
+        supervisor,
+        "api_command_while_streaming",
+        lambda *_args, **_kwargs: pytest.fail("invalid public state entered fallback"),
+    )
+
+    with pytest.raises(supervisor.ApiCommandError) as raised:
+        supervisor.manage_with_public_fallback(runtime, cached)
+
+    assert raised.value is failure
+
+
 def test_prepublic_api_failure_does_not_enter_public_fallback(load_script, monkeypatch):
     supervisor = load_script("youtube-autoencoder", "yta_prepublic_api_failure")
     runtime = supervisor.StreamRuntime(