[codex] Add generic YouTube AutoEncoder

[sumitake]

Summary:

• Add a generic headless FFmpeg supervisor with YouTube Live lifecycle management.
• Add a YouTube API helper for OAuth, reusable streams, broadcasts, transitions, and visible test-pattern runs.
• Add sample env config, systemd units, Raspberry Pi deployment notes, and secret-focused ignores.

Validation:

• python3 -m py_compile bin/youtube-autoencoder bin/youtube-autoencoder-api bin/youtube-autoencoder-test-pattern
• git diff --cached --check
• bin/youtube-autoencoder-api --help
• bin/youtube-autoencoder-api prepare-broadcast --help
• rg scans for SLORC/local-device labels and obvious committed secret patterns

Visibility:

• Repository remains private.

[gemini-code-assist]

Code Review

This pull request introduces YouTube AutoEncoder, a headless live-stream bridge that manages the YouTube Live broadcast lifecycle using FFmpeg and systemd. It includes the supervisor script, an API controller, a test pattern generator, configuration templates, and systemd service files. The review feedback identifies several critical issues, including potential subprocess pipe deadlocks in both the supervisor and test pattern execution when reading stdout, crashes when the OBS service file is missing, a security race condition during secret file creation, and non-interruptible sleep loops that hinder responsive shutdown and cancellation during restart delays, authorization, and stream-active waiting.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Subprocess Pipe Deadlock

When YTA_YOUTUBE_LIFECYCLE is enabled, the script starts the FFmpeg process with stdout=subprocess.PIPE and then blocks on wait_stream_active (which can take up to 180 seconds) and other API transitions before reading from the pipe.

During this time, FFmpeg is actively writing log output to the pipe. If the OS pipe buffer (typically 64KB on Linux) fills up, FFmpeg will block on its write calls, causing the stream ingest to freeze and the active check to eventually time out.

Using a background thread to continuously consume and log the output of the child process completely avoids this deadlock risk and simplifies the main loop.

    import threading
    def log_reader():
        if child.stdout is not None:
            for line in child.stdout:
                log("ffmpeg: " + redact_text(line.rstrip(), [input_url, ingest_url]))
    reader_thread = threading.Thread(target=log_reader, daemon=True)
    reader_thread.start()
    try:
        if lifecycle_enabled():
            wait_stream_active(stream_id)
            transition_broadcast(broadcast_id, "testing")
            testing_delay = int(float(env("YTA_YOUTUBE_TESTING_DELAY_SEC", "8")))
            sleep_end = time.monotonic() + testing_delay
            while time.monotonic() < sleep_end and not stopping:
                time.sleep(0.1)
            transition_broadcast(broadcast_id, "live")
        while child.poll() is None:
            if stopping:
                break
            if max_runtime and time.monotonic() - started > max_runtime:
                log(f"max runtime {max_runtime:.0f}s reached; rotating ffmpeg")
                child.send_signal(signal.SIGINT)
                break
            time.sleep(1)
        try:
            rc = child.wait(timeout=20)
            reader_thread.join(timeout=5)
            return rc
        except subprocess.TimeoutExpired:
            log("ffmpeg did not stop cleanly; killing")
            child.kill()
            return child.wait(timeout=10)

[gemini-code-assist]

OBS Service File Missing Bug

If OBS_SERVICE_FILE does not exist, obs_stream_name() raises FileNotFoundError. In ensure_stream(create_if_missing=True), this exception is not caught, causing the script to crash instead of creating the stream and the file.

Catching FileNotFoundError and ValueError in find_obs_stream and raising LookupError allows ensure_stream to catch it and proceed with creation.

Suggested change

	def find_obs_stream() -> dict[str, Any]:
	stream_name = obs_stream_name()
	for stream in list_streams():
	ingestion = ((stream.get("cdn") or {}).get("ingestionInfo") or {})
	if ingestion.get("streamName") == stream_name:
	return stream
	raise LookupError("no YouTube liveStream matched the configured OBS stream key")
	def find_obs_stream() -> dict[str, Any]:
	try:
	stream_name = obs_stream_name()
	except (FileNotFoundError, ValueError) as exc:
	raise LookupError("OBS stream key not configured") from exc
	for stream in list_streams():
	ingestion = ((stream.get("cdn") or {}).get("ingestionInfo") or {})
	if ingestion.get("streamName") == stream_name:
	return stream
	raise LookupError("no YouTube liveStream matched the configured OBS stream key")

[gemini-code-assist]

OBS Service File Missing Bug (Part 2)

When creating a stream, save_stream_to_obs attempts to read OBS_SERVICE_FILE. If the file does not exist, it will crash with FileNotFoundError.

Handling FileNotFoundError gracefully by defaulting to an empty dictionary allows the file to be created successfully.

Suggested change

	def save_stream_to_obs(stream: dict[str, Any]) -> None:
	ingestion = ((stream.get("cdn") or {}).get("ingestionInfo") or {})
	stream_name = ingestion.get("streamName")
	server = ingestion.get("rtmpsIngestionAddress") or ingestion.get("ingestionAddress")
	if not stream_name or not server:
	raise ValueError("created stream did not return ingestion URL and stream name")
	service = read_json(OBS_SERVICE_FILE)
	service.setdefault("settings", {})
	service["settings"]["server"] = server
	service["settings"]["key"] = stream_name
	write_secret_json(OBS_SERVICE_FILE, service)
	def save_stream_to_obs(stream: dict[str, Any]) -> None:
	ingestion = ((stream.get("cdn") or {}).get("ingestionInfo") or {})
	stream_name = ingestion.get("streamName")
	server = ingestion.get("rtmpsIngestionAddress") or ingestion.get("ingestionAddress")
	if not stream_name or not server:
	raise ValueError("created stream did not return ingestion URL and stream name")
	try:
	service = read_json(OBS_SERVICE_FILE)
	except FileNotFoundError:
	service = {}
	service.setdefault("settings", {})
	service["settings"]["server"] = server
	service["settings"]["key"] = stream_name
	write_secret_json(OBS_SERVICE_FILE, service)

[gemini-code-assist]

Subprocess Pipe Deadlock in Test Pattern

Just like in the main supervisor, run_visible_test starts the test pattern process with stdout=subprocess.PIPE and then blocks on wait_stream_active (up to 120 seconds) before reading from the pipe.

Using a background thread to continuously consume and print the output of the child process completely avoids this deadlock risk.

    import threading
    def log_reader():
        if child.stdout is not None:
            for line in child.stdout:
                print(line.rstrip(), flush=True)
    reader_thread = threading.Thread(target=log_reader, daemon=True)
    reader_thread.start()
    try:
        wait_stream_active(stream_id, args.wait_stream_active)
        transitioned = transition(broadcast_id, "testing")
        log(f"transitioned broadcast to testing status={(transitioned.get('status') or {}).get('lifeCycleStatus')}")
        sleep_end = time.monotonic() + args.testing_delay
        while time.monotonic() < sleep_end and not stopping:
            time.sleep(0.1)
        transitioned = transition(broadcast_id, "live")
        log(f"transitioned broadcast to live status={(transitioned.get('status') or {}).get('lifeCycleStatus')}")
        while child.poll() is None and not stopping:
            time.sleep(1)
        rc = child.wait(timeout=20)
        reader_thread.join(timeout=5)
        if args.complete:
            try:
                transition(broadcast_id, "complete")
                log("completed YouTube broadcast")
            except Exception as exc:  # noqa: BLE001
                log(f"could not complete broadcast: {exc}")
        return rc

[gemini-code-assist]

Non-Interruptible Restart Delay

Using time.sleep(delay) blocks the supervisor process from shutting down promptly when systemd sends a SIGTERM or when the user presses Ctrl+C. The process will hang until the sleep duration completes.

Replacing this with an interruptible sleep loop makes the service stop instantly.

Suggested change

	if not stopping:
	log(f"restarting in {delay:.0f}s")
	time.sleep(delay)
	if not stopping:
	log(f"restarting in {delay:.0f}s")
	sleep_end = time.monotonic() + delay
	while time.monotonic() < sleep_end and not stopping:
	time.sleep(0.1)

[gemini-code-assist]

Secure Secret File Creation

Creating the temporary file with default permissions and then calling os.chmod creates a brief race condition where other local users could read the OAuth refresh tokens or stream keys.

Using os.open with 0o600 ensures the file is created securely from the very beginning.

Suggested change

	def write_secret_json(path: pathlib.Path, data: dict[str, Any]) -> None:
	path.parent.mkdir(parents=True, exist_ok=True)
	tmp = path.with_suffix(path.suffix + ".tmp")
	with tmp.open("w", encoding="utf-8") as handle:
	json.dump(data, handle, indent=2, sort_keys=True)
	handle.write("\n")
	os.chmod(tmp, 0o600)
	tmp.replace(path)
	os.chmod(path, 0o600)
	def write_secret_json(path: pathlib.Path, data: dict[str, Any]) -> None:
	path.parent.mkdir(parents=True, exist_ok=True)
	tmp = path.with_suffix(path.suffix + ".tmp")
	fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
	with open(fd, "w", encoding="utf-8") as handle:
	json.dump(data, handle, indent=2, sort_keys=True)
	handle.write("\n")
	tmp.replace(path)

[gemini-code-assist]

Un-interruptible Authorization Loop

Because signal.SIGINT is overridden globally to set stopping = True without raising KeyboardInterrupt, the authorize loop cannot be cancelled with Ctrl+C because it never checks stopping.

Checking stopping and sleeping in small increments makes the command responsive to cancellation.

    while time.monotonic() < expires_at and not stopping:
        sleep_end = time.monotonic() + interval
        while time.monotonic() < sleep_end and not stopping:
            time.sleep(0.1)
        if stopping:
            print("Authorization cancelled.")
            return 1
        try:

[gemini-code-assist]

Un-interruptible Wait Loop

Similar to the authorization loop, wait_stream_active blocks on time.sleep(5) and does not check stopping, making it unresponsive to Ctrl+C or service stop signals.

Checking stopping and sleeping in small increments makes the wait loop perfectly responsive.

    while time.monotonic() < deadline and not stopping:
        stream = stream_by_id(stream_id)
        stream_status = (stream.get("status") or {}).get("streamStatus")
        if stream_status == "active":
            return stream
        if stream_status != last:
            log(f"YouTube stream status={stream_status}")
            last = stream_status
        sleep_end = time.monotonic() + 5
        while time.monotonic() < sleep_end and not stopping:
            time.sleep(0.1)
    if stopping:
        raise InterruptedError("cancelled while waiting for stream to become active")
    raise TimeoutError(f"stream {stream_id} did not become active within {timeout}s")