bot_recorder.py

#!/usr/bin/env python3
"""
bot_recorder.py

Headless Daily bot that joins a room, records each participant's audio and
video, and produces a single composed MP4 with dynamic layout switching.

Layout behaviour
────────────────
  The final video reflects exactly who was present at each moment:

  • 1 participant present  → full-width (OUTPUT_WIDTH × OUTPUT_HEIGHT)
  • 2 participants present → side-by-side hstack (each tile OUTPUT_WIDTH/2 wide)
  • 3+ participants        → equal-width hstack across all active tiles

  Layout switches at the exact second each participant joins or leaves.
  No freeze-frames, no black panels — clean cuts between segments.

  Example (Aditya T=5–65s, iPhone T=15–45s, session ends T=65s):
    T=5  → T=15  : Aditya full-width
    T=15 → T=45  : Aditya | iPhone  (side-by-side)
    T=45 → T=65  : Aditya full-width

Output directory layout
───────────────────────
  recordings/<session_id>/
    Aditya_video.mp4          ← per-participant encoded video (always kept)
    Aditya_audio.wav          ← per-participant audio WAV    (always kept)
    iPhone_video.mp4
    iPhone_audio.wav
    mixed.wav                 ← merged stereo audio
    session.json              ← timeline metadata (join/leave times, paths)
    seg_0000.mp4              ← intermediate segment files (deleted after compose)
    seg_0001.mp4
    <session_id>_recording.mp4 ← final composed output

  The recordings/ directory is NEVER deleted so you can re-run compose_video.py
  against the saved tracks if composition fails.

Usage
─────
  python bot_recorder.py --room-url "https://yourapp.daily.co/my-room"
  python bot_recorder.py --room-url "https://yourapp.daily.co/my-room" --dry-run

Requirements
────────────
  pip install daily-python pillow boto3 python-dotenv
  apt install ffmpeg   # or: brew install ffmpeg

Environment (.env or shell)
───────────────────────────
  AWS_ACCESS_KEY_ID
  AWS_SECRET_ACCESS_KEY
  AWS_REGION       (default: ap-south-1)
  S3_BUCKET
  BASE_CDN_URL     (optional — for logging the public CDN URL after upload)
  OUTPUT_WIDTH     (default: 1280)
  OUTPUT_HEIGHT    (default: 720)
  RECORDINGS_DIR   (default: ./recordings)
"""

import argparse
import json
import os
import re
import shutil
import sys
import wave
import threading
import time
import subprocess
from collections import defaultdict
from datetime import datetime
from typing import Optional

import boto3
from dotenv import load_dotenv
from daily import Daily, CallClient, EventHandler
from PIL import Image
from loguru import logger

load_dotenv()

# ── Logging ────────────────────────────────────────────────────────────────────

logger.remove()
logger.add(sys.stderr, level=os.environ.get("LOG_LEVEL", "INFO"))

# ── Constants ──────────────────────────────────────────────────────────────────

SAMPLE_RATE = 16000
NUM_CHANNELS = 1
SAMPLE_WIDTH = 2  # 16-bit PCM

S3_BUCKET = os.environ.get("S3_BUCKET")
AWS_REGION = os.environ.get("AWS_REGION", "ap-south-1")
BASE_CDN_URL = os.environ.get("BASE_CDN_URL", "")

OUTPUT_WIDTH = int(os.environ.get("OUTPUT_WIDTH", 1280))
OUTPUT_HEIGHT = int(os.environ.get("OUTPUT_HEIGHT", 720))

# Root directory for all session recordings — never deleted
RECORDINGS_DIR = os.environ.get("RECORDINGS_DIR", "./recordings")

DRY_RUN_SECS = 15
WRAP_UP_GRACE = 5


# ── Helpers ────────────────────────────────────────────────────────────────────


def safe_name(name: str) -> str:
    """Sanitise a participant name for use in file names."""
    return re.sub(r"[^\w\-]", "_", name).strip("_") or "participant"


# ── Per-participant recorder ───────────────────────────────────────────────────


class ParticipantRecorder:
    """
    Captures one participant's audio (raw PCM) and video (RGBA PNG frames).

    Video frames are saved as PNG files named by elapsed-microseconds so that
    ffmpeg's concat demuxer can reconstruct exact variable-rate timing without
    any fixed-fps assumptions.

    join_time_s / leave_time_s are wall-clock seconds relative to the moment
    the bot joined the room (set by RecorderBot).

    All output files live in session_dir (persistent — never deleted).
    Intermediate PNG frames live in a _frames/ subdir inside session_dir.
    """

    def __init__(self, participant_id: str, name: str, session_dir: str):
        self.participant_id = participant_id
        self.name = name
        self._safe = safe_name(name)
        self._session_dir = session_dir

        # Timestamps relative to session start — set by RecorderBot
        self.join_time_s: float = 0.0
        self.leave_time_s: Optional[float] = None

        # Audio
        self._audio_chunks: list[bytes] = []
        self._audio_lock = threading.Lock()

        # Video frames — PNGs in a subdirectory, named by elapsed microseconds
        self._frames_dir = os.path.join(session_dir, f"{self._safe}_frames")
        os.makedirs(self._frames_dir, exist_ok=True)
        self._frame_count = 0
        self._first_ts_us: Optional[int] = None

        # Final output paths (both persistent)
        self.raw_video_path = os.path.join(session_dir, f"{self._safe}_video.mp4")
        self.wav_path = os.path.join(session_dir, f"{self._safe}_audio.wav")

    # ── Video ──────────────────────────────────────────────────────────────────

    def write_video_frame(
        self, rgba_buffer: bytes, width: int, height: int, timestamp_us: int
    ):
        """Save one RGBA frame as a PNG. Filename = elapsed microseconds."""
        if self._first_ts_us is None:
            self._first_ts_us = timestamp_us

        elapsed_us = timestamp_us - self._first_ts_us
        fname = os.path.join(self._frames_dir, f"{elapsed_us:015d}.png")
        Image.frombytes("RGBA", (width, height), rgba_buffer).convert("RGB").save(fname)
        self._frame_count += 1

    def encode_video(self) -> bool:
        """
        Encode saved PNG frames → raw_video_path via ffmpeg concat demuxer.
        Each frame's display duration = gap to next frame's timestamp.
        Returns True on success.
        """
        logger.info("Encoding video for '{}'", self.name)
        logger.debug("frames_dir  : {}", self._frames_dir)
        logger.debug("frame_count : {}", self._frame_count)
        logger.debug("output      : {}", self.raw_video_path)

        if self._frame_count == 0:
            logger.info("→ SKIP: no frames captured")
            return False

        frame_files = sorted(
            f for f in os.listdir(self._frames_dir) if f.endswith(".png")
        )
        if not frame_files:
            logger.info("→ SKIP: no PNG files found in frames_dir")
            return False

        logger.debug("first frame : {}", frame_files[0])
        logger.debug("last frame  : {}", frame_files[-1])

        timestamps_us = [int(f[:-4]) for f in frame_files]
        total_duration_s = timestamps_us[-1] / 1_000_000 if timestamps_us else 0
        logger.debug("span        : {:.3f}s", total_duration_s)

        concat_path = os.path.join(self._frames_dir, "concat.txt")
        with open(concat_path, "w") as fh:
            for i, (fname, ts_us) in enumerate(zip(frame_files, timestamps_us)):
                full_path = os.path.join(self._frames_dir, fname)
                if i < len(timestamps_us) - 1:
                    duration_s = (timestamps_us[i + 1] - ts_us) / 1_000_000
                else:
                    duration_s = 1 / 30
                fh.write(f"file '{full_path}'\n")
                fh.write(f"duration {duration_s:.6f}\n")

        cmd = [
            "ffmpeg",
            "-y",
            "-f",
            "concat",
            "-safe",
            "0",
            "-i",
            concat_path,
            "-c:v",
            "libx264",
            "-preset",
            "fast",
            "-crf",
            "18",
            "-pix_fmt",
            "yuv420p",
            "-movflags",
            "+faststart",
            self.raw_video_path,
        ]
        r = subprocess.run(cmd, capture_output=True, text=True)
        if r.returncode == 0:
            mb = os.path.getsize(self.raw_video_path) / 1024 / 1024
            logger.info("→ encoded ✓  {:.1f} MB  {}", mb, self.raw_video_path)
            return True
        else:
            logger.error("→ FAILED:\n{}", (r.stderr or "")[-500:])
            return False

    # ── Audio ──────────────────────────────────────────────────────────────────

    def write_audio(self, pcm_bytes: bytes):
        with self._audio_lock:
            self._audio_chunks.append(pcm_bytes)

    def flush_audio_wav(self) -> bool:
        """Write accumulated PCM to wav_path. Returns True if anything was written."""
        pcm = self.get_audio_pcm()
        if not pcm:
            logger.info("'{}': no audio PCM — skipping WAV", self.name)
            return False
        with wave.open(self.wav_path, "wb") as wf:
            wf.setnchannels(NUM_CHANNELS)
            wf.setsampwidth(SAMPLE_WIDTH)
            wf.setframerate(SAMPLE_RATE)
            wf.writeframes(pcm)
        dur = len(pcm) / (SAMPLE_RATE * NUM_CHANNELS * SAMPLE_WIDTH)
        logger.info("'{}': {:.2f}s audio → {}", self.name, dur, self.wav_path)
        return True

    def get_audio_pcm(self) -> bytes:
        with self._audio_lock:
            return b"".join(self._audio_chunks)


# ── Timeline builder ───────────────────────────────────────────────────────────


class TimelineWindow:
    def __init__(self, start_s: float, end_s: float, recorders: list):
        self.start_s = start_s
        self.end_s = end_s
        self.recorders = recorders

    @property
    def duration_s(self) -> float:
        return self.end_s - self.start_s

    @property
    def n(self) -> int:
        return len(self.recorders)


def build_timeline(recorders: list) -> list:
    """
    Compute non-overlapping TimelineWindows from recorder join/leave times.

    Verbose: prints every breakpoint and the per-recorder evaluation for every
    candidate window so failures are immediately diagnosable.
    """
    logger.debug("Building timeline")
    logger.debug("Recorders ({}):", len(recorders))
    for r in recorders:
        logger.debug(
            "'{}' join={:.3f}s leave={} video_exists={} path={}",
            r.name,
            r.join_time_s,
            r.leave_time_s,
            os.path.exists(r.raw_video_path),
            r.raw_video_path,
        )

    breakpoints: set[float] = set()
    for r in recorders:
        breakpoints.add(r.join_time_s)
        if r.leave_time_s is not None:
            breakpoints.add(r.leave_time_s)

    sorted_bp = sorted(breakpoints)
    logger.debug("Breakpoints: {}", [f"{b:.3f}s" for b in sorted_bp])

    windows: list[TimelineWindow] = []

    for i in range(len(sorted_bp) - 1):
        t_start = sorted_bp[i]
        t_end = sorted_bp[i + 1]
        dur = t_end - t_start
        logger.debug(
            "Candidate window [{:.3f}s → {:.3f}s] dur={:.3f}s", t_start, t_end, dur
        )

        active = []
        for r in recorders:
            joined_ok = r.join_time_s <= t_start
            left_ok = r.leave_time_s is not None and r.leave_time_s >= t_end
            exists_ok = os.path.exists(r.raw_video_path)
            included = joined_ok and left_ok and exists_ok
            logger.debug(
                "'{}': join_ok={}({:.3f}<={:.3f}) leave_ok={}({} >= {:.3f}) "
                "video_exists={} → {}",
                r.name,
                joined_ok,
                r.join_time_s,
                t_start,
                left_ok,
                r.leave_time_s,
                t_end,
                exists_ok,
                "INCLUDED" if included else "excluded",
            )
            if included:
                active.append(r)

        if not active:
            logger.debug("→ SKIP: no active recorders")
        elif dur <= 0.01:
            logger.debug("→ SKIP: duration too short ({:.4f}s)", dur)
        else:
            names = " | ".join(r.name for r in active)
            logger.debug("→ WINDOW ✓  [{}]", names)
            windows.append(TimelineWindow(t_start, t_end, active))

    logger.debug("Timeline result: {} window(s)", len(windows))
    return windows


# ── Segment composer ───────────────────────────────────────────────────────────


def compose_segment(
    window: TimelineWindow,
    seg_index: int,
    session_dir: str,
    output_width: int,
    output_height: int,
) -> Optional[str]:
    """
    Produce one segment MP4 for a TimelineWindow.

    Each recorder's video is trimmed to the portion that falls in this window:
        trim_start = max(0, window.start_s - rec.join_time_s)
        trim_end   = trim_start + window.duration_s

    Tiles are scaled to (output_width // N) × output_height with letterboxing.
    No audio — muxed at concat step.
    """
    seg_path = os.path.join(session_dir, f"seg_{seg_index:04d}.mp4")
    n = window.n
    tile_w = output_width // n
    tile_h = output_height

    names = " | ".join(r.name for r in window.recorders)
    logger.info(
        "[seg {:02d}] T={:.3f}s → {:.3f}s  dur={:.3f}s  [{}]  tile={}x{}",
        seg_index,
        window.start_s,
        window.end_s,
        window.duration_s,
        names,
        tile_w,
        tile_h,
    )

    cmd = ["ffmpeg", "-y"]
    for rec in window.recorders:
        cmd += ["-i", rec.raw_video_path]

    filter_parts: list[str] = []
    tile_labels: list[str] = []

    for idx, rec in enumerate(window.recorders):
        trim_start = max(0.0, window.start_s - rec.join_time_s)
        trim_end = trim_start + window.duration_s
        out_lbl = f"[v{idx}]"

        logger.debug(
            "[seg {:02d}] '{}' trim {:.3f}s → {:.3f}s (join_offset={:.3f}s)",
            seg_index,
            rec.name,
            trim_start,
            trim_end,
            rec.join_time_s,
        )

        filter_parts.append(
            f"[{idx}:v]"
            f"trim=start={trim_start:.6f}:end={trim_end:.6f},"
            f"setpts=PTS-STARTPTS,"
            f"scale={tile_w}:{tile_h}:force_original_aspect_ratio=decrease,"
            f"pad={tile_w}:{tile_h}:(ow-iw)/2:(oh-ih)/2"
            f"{out_lbl}"
        )
        tile_labels.append(out_lbl)

    if n == 1:
        filter_parts.append(f"{tile_labels[0]}copy[vout]")
    else:
        filter_parts.append(f"{''.join(tile_labels)}hstack=inputs={n}[vout]")

    cmd += [
        "-filter_complex",
        ";".join(filter_parts),
        "-map",
        "[vout]",
        "-c:v",
        "libx264",
        "-preset",
        "fast",
        "-crf",
        "18",
        "-pix_fmt",
        "yuv420p",
        "-an",
        seg_path,
    ]

    r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode == 0:
        mb = os.path.getsize(seg_path) / 1024 / 1024
        logger.info("[seg {:02d}]   → ✓  {:.1f} MB  {}", seg_index, mb, seg_path)
        return seg_path
    else:
        logger.error("[seg {:02d}]   → FAILED:\n{}", seg_index, (r.stderr or "")[-500:])
        return None


def concat_segments(
    seg_paths: list[str],
    audio_path: Optional[str],
    out: str,
    session_dir: str,
):
    """Concatenate segment MP4s (stream-copy) and mux in mixed audio."""
    if not seg_paths:
        logger.error("No segments to concatenate.")
        return

    concat_list = os.path.join(session_dir, "segments.txt")
    with open(concat_list, "w") as fh:
        for sp in seg_paths:
            fh.write(f"file '{sp}'\n")

    has_audio = audio_path and os.path.exists(audio_path)

    logger.info(
        "Concatenating {} segment(s) audio={}",
        len(seg_paths),
        "yes" if has_audio else "no",
    )

    cmd = [
        "ffmpeg",
        "-y",
        "-f",
        "concat",
        "-safe",
        "0",
        "-i",
        concat_list,
    ]
    if has_audio:
        cmd += ["-i", audio_path]

    cmd += ["-map", "0:v"]
    if has_audio:
        cmd += ["-map", "1:a"]

    cmd += [
        # Re-encode video instead of stream-copy (-c:v copy).
        # Stream-copy requires identical codec parameters across all segments.
        # If Daily adapts resolution mid-session (e.g. 640x360 → 320x180 on
        # network degradation), segments will have different dimensions and
        # stream-copy will fail with a hard error. Re-encoding handles this.
        "-c:v",
        "libx264",
        "-preset",
        "fast",
        "-crf",
        "18",
        "-pix_fmt",
        "yuv420p",
        "-c:a",
        "aac",
        "-b:a",
        "128k",
        "-movflags",
        "+faststart",
        out,
    ]

    r = subprocess.run(cmd, capture_output=True, text=True)
    if r.returncode == 0:
        mb = os.path.getsize(out) / 1024 / 1024
        logger.info("Final video ✓  {:.1f} MB → {}", mb, out)
    else:
        logger.error("Concat FAILED:\n{}", (r.stderr or "")[-500:])


# ── Bot ────────────────────────────────────────────────────────────────────────


class RecorderBot(EventHandler):
    """
    Headless Daily bot recorder.

    daily-python quirks
    ────────────────────
    • __new__ must be overridden — EventHandler.__new__ rejects kwargs.
    • CallClient must be created AFTER super().__init__().
    • Media callbacks must be direct methods on the class, not closures.
    """

    def __new__(cls, *args, **kwargs):
        return super().__new__(cls)

    def __init__(self, room_url: str, dry_run: bool = False):
        super().__init__()

        self.room_url = room_url
        self.dry_run = dry_run

        self._session_id = datetime.utcnow().strftime("%Y%m%d_%H%M%S")

        # Persistent session directory — never deleted
        self._session_dir = os.path.join(
            os.path.abspath(RECORDINGS_DIR), self._session_id
        )
        os.makedirs(self._session_dir, exist_ok=True)

        self._session_start_time: Optional[float] = None

        self._recorders: dict[str, ParticipantRecorder] = {}
        self._lock = threading.Lock()
        self._participant_count = 0
        self._done = threading.Event()
        self._wrap_timer: Optional[threading.Timer] = None

        # dry-run stats
        self._dry_audio_count: dict[str, int] = defaultdict(int)
        self._dry_audio_bytes: dict[str, int] = defaultdict(int)
        self._dry_video_count: dict[str, int] = defaultdict(int)
        self._dry_video_sizes: dict[str, set] = defaultdict(set)
        self._dry_video_ts: dict[str, list] = defaultdict(list)
        self._dry_first_audio: set = set()
        self._dry_first_video: set = set()

        self._client = CallClient(event_handler=self)

        logger.info("Mode        : {}", "DRY-RUN" if dry_run else "RECORDING")
        logger.info("Session     : {}", self._session_id)
        logger.info("Session dir : {}", self._session_dir)

    # ── Helpers ────────────────────────────────────────────────────────────────

    def _elapsed(self) -> float:
        if self._session_start_time is None:
            return 0.0
        return time.time() - self._session_start_time

    def _register_renderers(self, pid: str):
        """Register audio + video renderers for a participant."""
        self._client.set_audio_renderer(
            pid,
            callback=self.on_audio_frame,
            sample_rate=SAMPLE_RATE,
            callback_interval_ms=20,
        )
        self._client.set_video_renderer(
            pid,
            callback=self.on_video_frame,
            color_format="RGBA",
        )

    # ── EventHandler callbacks ─────────────────────────────────────────────────

    def on_joined(self, data, error):
        if error:
            logger.error("Join failed: {}", error)
            self._done.set()
            return

        logger.info("Joined room ✓  session_start={:.3f}", self._session_start_time)

        # Bootstrap recorders for participants already in the room when the bot joins.
        # Guard against duplicates — on_participant_joined may fire for the same pid
        # immediately after on_joined, so we only create if not already present.
        participants = data.get("participants", {})
        already_present = [p for p in participants if p != "local"]
        logger.info("Participants already present: {}", len(already_present))
        for pid, info in participants.items():
            if pid == "local":
                continue
            name = info.get("info", {}).get("userName", pid[:8])
            logger.info(
                "Already in room: '{}' ({}) — bootstrapping at T=0.0s",
                name,
                pid[:8],
            )
            if not self.dry_run:
                with self._lock:
                    if pid not in self._recorders:
                        rec = ParticipantRecorder(pid, name, self._session_dir)
                        rec.join_time_s = 0.0
                        self._recorders[pid] = rec
                        self._participant_count += 1
                        logger.info("recorder created join_time_s=0.000")
                    else:
                        logger.debug(
                            "recorder already exists for {} — skipping", pid[:8]
                        )
                        continue  # renderers already registered, skip
            self._register_renderers(pid)

        if self.dry_run:
            logger.info("Observing for {}s — join the room now.", DRY_RUN_SECS)
            threading.Timer(DRY_RUN_SECS, self._dry_run_report).start()

    def on_call_state_updated(self, state):
        if state == "joined" and self._session_start_time is None:
            self._session_start_time = time.time()
            logger.debug("session_start_time set: {:.3f}", self._session_start_time)

        logger.info("Call state → {}  T={:.2f}s", state, self._elapsed())

        if state == "left":
            logger.info("Call state 'left' — signalling done.")
            self._done.set()

    def on_participant_joined(self, participant):
        pid = participant["id"]
        name = participant.get("info", {}).get("userName", pid[:8])
        if pid == "local":
            return

        t = self._elapsed()
        logger.info("on_participant_joined: '{}' ({})  T={:.3f}s", name, pid[:8], t)

        is_new = False
        if not self.dry_run:
            with self._lock:
                if pid not in self._recorders:
                    rec = ParticipantRecorder(pid, name, self._session_dir)
                    rec.join_time_s = t
                    self._recorders[pid] = rec
                    self._participant_count += 1
                    is_new = True
                    logger.info("recorder created join_time_s={:.3f}", t)
                else:
                    # Already bootstrapped via on_joined — renderers already
                    # registered, do NOT register again (causes doubled frames).
                    logger.debug(
                        "recorder already exists for {} — skipping (bootstrapped)",
                        pid[:8],
                    )
                if self._wrap_timer:
                    self._wrap_timer.cancel()
                    self._wrap_timer = None
                    logger.debug("wrap-up timer cancelled (participant rejoined)")
        else:
            is_new = True  # always register renderers in dry-run

        # Only register renderers for genuinely new participants.
        # Registering twice causes doubled audio/video callbacks.
        if is_new:
            self._register_renderers(pid)

    def on_participant_left(self, participant, reason):
        pid = participant["id"]
        if pid == "local":
            return

        t = self._elapsed()
        with self._lock:
            rec = self._recorders.get(pid)
            name = rec.name if rec else pid[:8]

        logger.info(
            "on_participant_left: '{}' ({}) T={:.3f}s reason={}",
            name,
            pid[:8],
            t,
            reason,
        )

        if not self.dry_run:
            with self._lock:
                if rec:
                    rec.leave_time_s = t
                    logger.debug("leave_time_s set to {:.3f}s", t)
                else:
                    logger.warning("no recorder found for {}", pid[:8])
                self._participant_count = max(0, self._participant_count - 1)
                remaining = self._participant_count
                logger.debug("participants remaining: {}", remaining)

            if remaining <= 0:
                logger.info("scheduling wrap-up in {}s...", WRAP_UP_GRACE)
                self._wrap_timer = threading.Timer(WRAP_UP_GRACE, self._done.set)
                self._wrap_timer.start()

    def on_error(self, message):
        logger.error("Room error: {}", message)

    # ── Media callbacks ────────────────────────────────────────────────────────

    def on_audio_frame(self, participant_id, audio_data, audio_source):
        if self.dry_run:
            chunk = bytes(audio_data.audio_frames)
            self._dry_audio_count[participant_id] += 1
            self._dry_audio_bytes[participant_id] += len(chunk)
            if participant_id not in self._dry_first_audio:
                self._dry_first_audio.add(participant_id)
                logger.info(
                    "[dry-run] First audio {}: {} bytes | sample_rate={} | channels={} | bits={} | source={}",
                    participant_id[:8],
                    len(chunk),
                    audio_data.sample_rate,
                    audio_data.num_channels,
                    audio_data.bits_per_sample,
                    audio_source,
                )
        else:
            rec = self._recorders.get(participant_id)
            if rec:
                rec.write_audio(bytes(audio_data.audio_frames))

    def on_video_frame(self, participant_id, video_frame, video_source):
        if self.dry_run:
            self._dry_video_count[participant_id] += 1
            self._dry_video_sizes[participant_id].add(
                (video_frame.width, video_frame.height)
            )
            self._dry_video_ts[participant_id].append(video_frame.timestamp_us)
            if participant_id not in self._dry_first_video:
                self._dry_first_video.add(participant_id)
                buf = bytes(video_frame.buffer)
                expected = video_frame.width * video_frame.height * 4
                match = "✓" if len(buf) == expected else f"✗ MISMATCH (got {len(buf)})"
                logger.info(
                    "[dry-run] First video {}: {}x{} | buffer={} | expected={} | {} | timestamp_us={} | source={}",
                    participant_id[:8],
                    video_frame.width,
                    video_frame.height,
                    len(buf),
                    expected,
                    match,
                    video_frame.timestamp_us,
                    video_source,
                )
        else:
            rec = self._recorders.get(participant_id)
            if rec:
                rec.write_video_frame(
                    bytes(video_frame.buffer),
                    video_frame.width,
                    video_frame.height,
                    video_frame.timestamp_us,
                )

    # ── Dry-run report ─────────────────────────────────────────────────────────

    def _dry_run_report(self):
        logger.info("=" * 60)
        logger.info("[dry-run] OBSERVATION REPORT")
        logger.info("=" * 60)
        all_pids = set(self._dry_audio_count) | set(self._dry_video_count)
        if not all_pids:
            logger.info("[dry-run] No participants observed.")
        else:
            for pid in sorted(all_pids):
                logger.info("Participant: {}...", pid[:8])
                a_bytes = self._dry_audio_bytes[pid]
                a_dur = a_bytes / (SAMPLE_RATE * NUM_CHANNELS * SAMPLE_WIDTH)
                logger.info(
                    "Audio  : {:,} callbacks | {:,} bytes | ~{:.1f}s",
                    self._dry_audio_count[pid],
                    a_bytes,
                    a_dur,
                )
                v_count = self._dry_video_count[pid]
                ts_list = self._dry_video_ts[pid]
                if len(ts_list) >= 2:
                    span_s = (ts_list[-1] - ts_list[0]) / 1_000_000
                    avg_fps = (len(ts_list) - 1) / span_s if span_s > 0 else 0
                    fps_str = f"~{avg_fps:.1f} fps"
                else:
                    fps_str = f"~{v_count / DRY_RUN_SECS:.1f} fps (wall-clock est.)"
                logger.info(
                    "Video  : {:,} callbacks | {} | resolutions={}",
                    v_count,
                    fps_str,
                    self._dry_video_sizes.get(pid),
                )
        logger.info("Audio received: {}", "✓" if self._dry_audio_count else "✗ none")
        logger.info("Video received: {}", "✓" if self._dry_video_count else "✗ none")
        logger.info("=" * 60)
        self._done.set()

    # ── Lifecycle ──────────────────────────────────────────────────────────────

    def run(self):
        logger.info("Joining {} ...", self.room_url)
        self._client.join(
            self.room_url,
            client_settings={
                "inputs": {
                    "camera": {"isEnabled": False},
                    "microphone": {"isEnabled": False},
                }
            },
        )
        try:
            self._done.wait()
        except KeyboardInterrupt:
            logger.info("Interrupted by user.")
        finally:
            if self._wrap_timer:
                self._wrap_timer.cancel()
            logger.info("Leaving room...")
            self._client.leave()
            time.sleep(1)
            self._client.release()
            Daily.deinit()

            if not self.dry_run:
                self._finalise()
            else:
                logger.info("Dry-run complete — no files written.")

    # ── Post-processing ────────────────────────────────────────────────────────

    def _finalise(self):
        recorders = list(self._recorders.values())

        logger.info("POST-PROCESSING session={}", self._session_id)
        logger.info("session_dir : {}", self._session_dir)
        logger.info("recorders   : {}", len(recorders))

        if not recorders:
            logger.info("Nothing recorded.")
            return

        # Seal participants with no leave event
        session_end_s = self._elapsed()
        logger.info("session_end_s = {:.3f}s", session_end_s)
        for rec in recorders:
            if rec.leave_time_s is None:
                rec.leave_time_s = session_end_s
                logger.info(
                    "'{}' had no leave event — sealed at T={:.3f}s",
                    rec.name,
                    session_end_s,
                )

        # ── Step 1: Encode per-participant videos ──────────────────────────────
        logger.info("Step 1: Encoding per-participant videos")
        for rec in recorders:
            rec.encode_video()

        # ── Step 2: Write per-participant WAV files ────────────────────────────
        logger.info("Step 2: Writing audio WAVs")
        audio_paths: list[str] = []
        for rec in recorders:
            if rec.flush_audio_wav():
                audio_paths.append(rec.wav_path)

        # ── Step 3: Mix audio ──────────────────────────────────────────────────
        logger.info("Step 3: Mixing audio")
        mixed_audio = os.path.join(self._session_dir, "mixed.wav")
        self._mix_audio(audio_paths, mixed_audio)

        # ── Step 4: Write session.json (metadata for compose_video.py) ────────
        logger.info("Step 4: Writing session metadata")
        session_meta = {
            "session_id": self._session_id,
            "session_dir": self._session_dir,
            "participants": [
                {
                    "name": rec.name,
                    "pid": rec.participant_id,
                    "join_time_s": rec.join_time_s,
                    "leave_time_s": rec.leave_time_s,
                    "video_path": rec.raw_video_path,
                    "audio_path": (
                        rec.wav_path if os.path.exists(rec.wav_path) else None
                    ),
                }
                for rec in recorders
            ],
        }
        meta_path = os.path.join(self._session_dir, "session.json")
        with open(meta_path, "w") as fh:
            json.dump(session_meta, fh, indent=2)
        logger.info("session.json → {}", meta_path)
        logger.debug("session.json contents:\n{}", json.dumps(session_meta, indent=2))

        # ── Step 5: Build timeline ─────────────────────────────────────────────
        logger.info("Step 5: Building timeline")
        video_recorders = [r for r in recorders if os.path.exists(r.raw_video_path)]
        logger.info(
            "Recorders with video files: {}/{}",
            len(video_recorders),
            len(recorders),
        )
        for r in recorders:
            exists = os.path.exists(r.raw_video_path)
            logger.info("'{}': {} exists={}", r.name, r.raw_video_path, exists)

        if not video_recorders:
            logger.error("No video files produced — aborting compose.")
            return

        windows = build_timeline(video_recorders)

        if not windows:
            logger.error("Timeline produced no windows — aborting compose.")
            logger.info(
                "To retry later: python compose_video.py --session {}", meta_path
            )
            return

        logger.info("Timeline summary: {} window(s)", len(windows))
        for w in windows:
            names = " | ".join(r.name for r in w.recorders)
            logger.info("T={:.3f}s → T={:.3f}s  [{}]", w.start_s, w.end_s, names)

        # ── Step 6: Encode segments ────────────────────────────────────────────
        logger.info("Step 6: Encoding segments")
        seg_paths: list[str] = []
        for i, w in enumerate(windows):
            seg = compose_segment(w, i, self._session_dir, OUTPUT_WIDTH, OUTPUT_HEIGHT)
            if seg:
                seg_paths.append(seg)

        if not seg_paths:
            logger.error("All segments failed — aborting.")
            return

        # ── Step 7: Concatenate + mux audio ───────────────────────────────────
        logger.info("Step 7: Concatenating segments")
        out_name = f"{self._session_id}_recording.mp4"
        out_path = os.path.join(self._session_dir, out_name)
        concat_segments(seg_paths, mixed_audio, out_path, self._session_dir)

        # Clean up segment files (keep everything else)
        for sp in seg_paths:
            try:
                os.remove(sp)
            except OSError:
                pass
        seg_list = os.path.join(self._session_dir, "segments.txt")
        if os.path.exists(seg_list):
            os.remove(seg_list)

        # ── Step 8: Upload ─────────────────────────────────────────────────────
        logger.info("Step 8: Uploading")
        self._upload(out_path, out_name)

        logger.info("DONE")
        logger.info("All files preserved at: {}", self._session_dir)
        logger.info(
            "To re-run composition: python compose_video.py --session {}",
            meta_path,
        )

    def _mix_audio(self, paths: list[str], out: str):
        if not paths:
            logger.info("No audio to mix.")
            return

        recorders = list(self._recorders.values())
        earliest_join = min(r.join_time_s for r in recorders)
        logger.info("Audio mix: earliest_join={:.3f}s (video T=0)", earliest_join)

        if len(paths) == 1:
            rec = next((r for r in recorders if r.wav_path == paths[0]), None)
            pad_s = (rec.join_time_s - earliest_join) if rec else 0.0
            if pad_s > 0.001:
                padded = os.path.join(
                    self._session_dir, f"{rec._safe}_audio_padded.wav"
                )
                self._pad_wav(paths[0], padded, pad_s)
                shutil.copy(padded, out)
            else:
                shutil.copy(paths[0], out)
            logger.info("Single audio track → {}", out)
            return

        padded_paths = []
        for rec in recorders:
            if not os.path.exists(rec.wav_path):
                continue
            pad_s = rec.join_time_s - earliest_join
            logger.debug(
                "'{}': pad={:.3f}s join={:.3f}s", rec.name, pad_s, rec.join_time_s
            )
            if pad_s > 0.001:
                padded = os.path.join(
                    self._session_dir, f"{rec._safe}_audio_padded.wav"
                )
                self._pad_wav(rec.wav_path, padded, pad_s)
                padded_paths.append(padded)
            else:
                padded_paths.append(rec.wav_path)

        cmd = ["ffmpeg", "-y"]
        for p in padded_paths:
            cmd += ["-i", p]
        cmd += [
            "-filter_complex",
            f"amix=inputs={len(padded_paths)}:duration=longest:normalize=0",
            out,
        ]
        r = subprocess.run(cmd, capture_output=True, text=True)
        if r.returncode == 0:
            logger.info("Audio mixed ✓ → {}", out)
        else:
            logger.warning("Audio mix warning:\n{}", (r.stderr or "")[-300:])

    def _pad_wav(self, src: str, dst: str, pad_s: float):
        """Prepend pad_s seconds of silence to src WAV, write to dst."""
        delay_ms = int(pad_s * 1000)
        cmd = [
            "ffmpeg",
            "-y",
            "-i",
            src,
            "-af",
            f"adelay={delay_ms}:all=1",
            dst,
        ]
        r = subprocess.run(cmd, capture_output=True, text=True)
        if r.returncode != 0:
            logger.warning(
                "WAV pad warning — falling back to unpadded:\n{}",
                (r.stderr or "")[-200:],
            )
            shutil.copy(src, dst)

    def _upload(self, path: str, name: str):
        if not os.path.exists(path):
            logger.error("Output file missing — skipping upload.")
            return
        if not S3_BUCKET:
            logger.warning("S3_BUCKET not set — file at: {}", path)
            return
        key = f"recordings/{self._session_id}/{name}"
        logger.info("Uploading → s3://{}/{}", S3_BUCKET, key)
        try:
            boto3.client("s3", region_name=AWS_REGION).upload_file(
                path,
                S3_BUCKET,
                key,
                ExtraArgs={"ContentType": "video/mp4"},
            )
            cdn_url = (
                f"{BASE_CDN_URL}/{key}" if BASE_CDN_URL else f"s3://{S3_BUCKET}/{key}"
            )
            logger.info("Uploaded ✓  {}", cdn_url)
        except Exception as exc:
            logger.exception("Upload failed. File at: {}", path)


# ── Entry point ────────────────────────────────────────────────────────────────


def main():
    parser = argparse.ArgumentParser(
        description=(
            "Headless Daily bot recorder — records all participants and "
            "produces a composed MP4 with dynamic layout switching."
        )
    )
    parser.add_argument(
        "--room-url",
        required=True,
        help="Full Daily room URL, e.g. https://yourapp.daily.co/my-room",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help=f"Observe for {DRY_RUN_SECS}s and print what arrives — no files written.",
    )
    args = parser.parse_args()

    Daily.init()
    bot = RecorderBot(room_url=args.room_url, dry_run=args.dry_run)
    bot.run()


if __name__ == "__main__":
    main()