from __future__ import annotations

import asyncio
import hashlib
import logging
import os
import queue
import re
import threading
import time
import uuid
import wave
import weakref
from collections import deque
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Mapping

from app.utils.logging import log_flow_event
from app.utils.stt_audio_upload import (
    OCIUploadConfig,
    _OCI_UPLOAD_LOCK,
    _oci_client,
    _oci_upload_config_from_env,
    _reset_oci_client,
)

logger = logging.getLogger(__name__)

_DEFAULT_QUEUE_SIZE = 1024
_DEFAULT_MAX_CLIENT_BUFFER_FRAMES = 512
_UPLOAD_QUEUE_SIZE = 8
_UPLOAD_MAX_ATTEMPTS = 3
_UPLOAD_RETRY_BASE_DELAY_S = 0.25
_MAX_PART_CHARS = 64
_SAFE_PART_RE = re.compile(r"[^A-Za-z0-9_.-]+")
_SENTINEL = object()
_RESERVED_LOG_FIELD_NAMES = frozenset(
    {
        "audio_duration_ms",
        "bytes",
        "duration_ms",
        "dropped_client_buffer_frames",
        "dropped_queue_frames",
        "error",
        "frame_ms",
        "frames",
        "object_name",
        "path",
        "reason",
        "sample_rate",
        "session_id",
    }
)


@dataclass(frozen=True, slots=True)
class EntireCallRecordingResult:
    object_name: str
    path: Path
    bytes: int
    frames: int
    duration_ms: int
    dropped_queue_frames: int
    dropped_client_buffer_frames: int
    upload_enqueued: bool


@dataclass(frozen=True, slots=True)
class _RecordingEvent:
    source: str
    frame: bytes


@dataclass(frozen=True, slots=True)
class EntireCallUploadItem:
    object_name: str
    path: Path
    session_id: str
    bytes: int
    duration_ms: int
    dropped_queue_frames: int
    dropped_client_buffer_frames: int
    metadata: Mapping[str, Any]
    logger: Any | None = None
    timeline: Any | None = None


def _safe_path_part(value: Any, *, fallback: str) -> str:
    raw = str(value or "").strip()
    if not raw:
        raw = fallback
    safe = _SAFE_PART_RE.sub("_", raw).strip("._-") or fallback
    if len(safe) <= _MAX_PART_CHARS:
        return safe

    digest = hashlib.sha1(safe.encode("utf-8")).hexdigest()[:12]
    prefix_len = max(1, _MAX_PART_CHARS - len(digest) - 1)
    return f"{safe[:prefix_len].rstrip('._-')}-{digest}"


def build_entire_call_object_name(
    *,
    session_id: Any,
    now: datetime | None = None,
) -> str:
    dt = now or datetime.now()
    session_part = _safe_path_part(session_id, fallback="unknown-session")
    return f"{dt:%Y-%m-%d}/entire_call/{session_part}.wav"


def _env_bool(name: str, default: bool) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


def _tmp_dir_from_env() -> Path:
    return Path(os.getenv("ENTIRE_CALL_RECORDING_TMP_DIR", "./recordings/entire_call_tmp"))


def _metadata_fields(metadata: Mapping[str, Any]) -> dict[str, Any]:
    return {
        str(key): value
        for key, value in metadata.items()
        if value not in (None, "") and str(key) not in _RESERVED_LOG_FIELD_NAMES
    }


def _emit_timeline(timeline: Any | None, event: str, **fields: Any) -> None:
    if timeline is None:
        return
    try:
        timeline.emit(event, **fields)
    except Exception:
        pass


def _cleanup_path(path: Path, active_logger: Any | None = None) -> None:
    try:
        path.unlink(missing_ok=True)
    except Exception:
        (active_logger or logger).debug(
            "[entire-call-recording] failed cleaning temp file path=%s",
            path,
            exc_info=True,
        )


def _put_object_file_sync(item: EntireCallUploadItem, config: OCIUploadConfig) -> Any:
    with _OCI_UPLOAD_LOCK:
        client = _oci_client(config)
        with item.path.open("rb") as body:
            return client.put_object(
                namespace_name=config.namespace,
                bucket_name=config.bucket,
                object_name=item.object_name,
                put_object_body=body,
            )


def _upload_max_attempts() -> int:
    raw = os.getenv("ENTIRE_CALL_RECORDING_UPLOAD_MAX_ATTEMPTS", str(_UPLOAD_MAX_ATTEMPTS))
    try:
        return max(1, int(raw))
    except (TypeError, ValueError):
        return _UPLOAD_MAX_ATTEMPTS


def _upload_retry_base_delay_s() -> float:
    raw = os.getenv(
        "ENTIRE_CALL_RECORDING_UPLOAD_RETRY_BASE_DELAY_S",
        str(_UPLOAD_RETRY_BASE_DELAY_S),
    )
    try:
        return max(0.0, float(raw))
    except (TypeError, ValueError):
        return _UPLOAD_RETRY_BASE_DELAY_S


async def _upload_item(item: EntireCallUploadItem) -> None:
    started = time.perf_counter()
    config = _oci_upload_config_from_env()
    if config is None:
        raise RuntimeError("OCI Object Storage upload config is missing")

    max_attempts = _upload_max_attempts()
    base_delay_s = _upload_retry_base_delay_s()
    response: Any | None = None
    for attempt in range(1, max_attempts + 1):
        try:
            response = await asyncio.to_thread(_put_object_file_sync, item, config)
            break
        except Exception as exc:
            # Do not close/reset the shared HTTP session while an STT upload is
            # using it. The same lock serializes reset and put_object calls.
            with _OCI_UPLOAD_LOCK:
                _reset_oci_client()
            if attempt >= max_attempts:
                raise
            delay_s = base_delay_s * (2 ** (attempt - 1))
            log_flow_event(
                item.logger or logger,
                "entire_call_recording_upload_retry",
                session_id=item.session_id,
                object_name=item.object_name,
                attempt=attempt,
                max_attempts=max_attempts,
                delay_ms=round(delay_s * 1000),
                error_type=type(exc).__name__,
                **_metadata_fields(item.metadata),
            )
            if delay_s:
                await asyncio.sleep(delay_s)

    if response is None:
        raise RuntimeError("OCI Object Storage upload returned no response")
    duration_ms = round((time.perf_counter() - started) * 1000)
    active_logger = item.logger or logger
    log_flow_event(
        active_logger,
        "entire_call_recording_upload_done",
        session_id=item.session_id,
        object_name=item.object_name,
        bytes=item.bytes,
        audio_duration_ms=item.duration_ms,
        duration_ms=duration_ms,
        bucket=config.bucket,
        namespace=config.namespace,
        oci_status=getattr(response, "status", None),
        dropped_queue_frames=item.dropped_queue_frames,
        dropped_client_buffer_frames=item.dropped_client_buffer_frames,
        **_metadata_fields(item.metadata),
    )
    _emit_timeline(
        item.timeline,
        "entire_call_recording_upload_done",
        session_id=item.session_id,
        object_name=item.object_name,
        bytes=item.bytes,
        audio_duration_ms=item.duration_ms,
        duration_ms=duration_ms,
        bucket=config.bucket,
        namespace=config.namespace,
        oci_status=getattr(response, "status", None),
        dropped_queue_frames=item.dropped_queue_frames,
        dropped_client_buffer_frames=item.dropped_client_buffer_frames,
        **_metadata_fields(item.metadata),
    )


class _EntireCallUploadWorker:
    def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
        self._loop = loop
        self._queue: asyncio.Queue[EntireCallUploadItem] = asyncio.Queue(maxsize=_UPLOAD_QUEUE_SIZE)
        self._task: asyncio.Task[None] | None = None

    def enqueue(self, item: EntireCallUploadItem) -> bool:
        if self._task is None or self._task.done():
            self._task = self._loop.create_task(
                self._run(),
                name="entire-call-recording-upload-worker",
            )
        try:
            self._queue.put_nowait(item)
            return True
        except asyncio.QueueFull:
            return False

    async def _run(self) -> None:
        while True:
            item = await self._queue.get()
            active_logger = item.logger or logger
            try:
                await _upload_item(item)
                _cleanup_path(item.path, active_logger)
            except Exception:
                log_flow_event(
                    active_logger,
                    "entire_call_recording_upload_failed",
                    session_id=item.session_id,
                    object_name=item.object_name,
                    bytes=item.bytes,
                    audio_duration_ms=item.duration_ms,
                    retained_path=str(item.path),
                    dropped_queue_frames=item.dropped_queue_frames,
                    dropped_client_buffer_frames=item.dropped_client_buffer_frames,
                    **_metadata_fields(item.metadata),
                )
                _emit_timeline(
                    item.timeline,
                    "entire_call_recording_upload_failed",
                    session_id=item.session_id,
                    object_name=item.object_name,
                    bytes=item.bytes,
                    audio_duration_ms=item.duration_ms,
                    retained_path=str(item.path),
                    dropped_queue_frames=item.dropped_queue_frames,
                    dropped_client_buffer_frames=item.dropped_client_buffer_frames,
                    **_metadata_fields(item.metadata),
                )
                active_logger.exception(
                    "[entire-call-recording][upload_failed] session_id=%s object_name=%s",
                    item.session_id,
                    item.object_name,
                )
            finally:
                self._queue.task_done()


_UPLOAD_WORKERS: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, _EntireCallUploadWorker] = (
    weakref.WeakKeyDictionary()
)
_UPLOAD_WORKERS_LOCK = threading.Lock()


def _worker_for_loop(loop: asyncio.AbstractEventLoop) -> _EntireCallUploadWorker:
    with _UPLOAD_WORKERS_LOCK:
        worker = _UPLOAD_WORKERS.get(loop)
        if worker is None:
            worker = _EntireCallUploadWorker(loop)
            _UPLOAD_WORKERS[loop] = worker
        return worker


def enqueue_entire_call_upload(
    *,
    path: Path,
    object_name: str,
    session_id: str,
    bytes: int,
    duration_ms: int,
    dropped_queue_frames: int = 0,
    dropped_client_buffer_frames: int = 0,
    metadata: Mapping[str, Any] | None = None,
    logger_override: Any | None = None,
    timeline: Any | None = None,
    cleanup_on_drop: bool = False,
) -> str | None:
    active_logger = logger_override or logger
    metadata = metadata or {}
    if _oci_upload_config_from_env() is None:
        log_flow_event(
            active_logger,
            "entire_call_recording_upload_drop",
            session_id=session_id,
            object_name=object_name,
            reason="missing_oci_config",
            bytes=bytes,
            audio_duration_ms=duration_ms,
            **_metadata_fields(metadata),
        )
        _emit_timeline(
            timeline,
            "entire_call_recording_upload_drop",
            session_id=session_id,
            object_name=object_name,
            reason="missing_oci_config",
            bytes=bytes,
            audio_duration_ms=duration_ms,
            **_metadata_fields(metadata),
        )
        if cleanup_on_drop:
            _cleanup_path(path, active_logger)
        return None

    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        log_flow_event(
            active_logger,
            "entire_call_recording_upload_drop",
            session_id=session_id,
            object_name=object_name,
            reason="no_running_loop",
            bytes=bytes,
            audio_duration_ms=duration_ms,
            **_metadata_fields(metadata),
        )
        if cleanup_on_drop:
            _cleanup_path(path, active_logger)
        return None

    item = EntireCallUploadItem(
        object_name=object_name,
        path=path,
        session_id=session_id,
        bytes=bytes,
        duration_ms=duration_ms,
        dropped_queue_frames=dropped_queue_frames,
        dropped_client_buffer_frames=dropped_client_buffer_frames,
        metadata=metadata,
        logger=active_logger,
        timeline=timeline,
    )
    queued = _worker_for_loop(loop).enqueue(item)
    if not queued:
        log_flow_event(
            active_logger,
            "entire_call_recording_upload_drop",
            session_id=session_id,
            object_name=object_name,
            reason="queue_full",
            bytes=bytes,
            audio_duration_ms=duration_ms,
            **_metadata_fields(metadata),
        )
        _emit_timeline(
            timeline,
            "entire_call_recording_upload_drop",
            session_id=session_id,
            object_name=object_name,
            reason="queue_full",
            bytes=bytes,
            audio_duration_ms=duration_ms,
            **_metadata_fields(metadata),
        )
        if cleanup_on_drop:
            _cleanup_path(path, active_logger)
        return object_name

    log_flow_event(
        active_logger,
        "entire_call_recording_upload_enqueued",
        session_id=session_id,
        object_name=object_name,
        bytes=bytes,
        audio_duration_ms=duration_ms,
        dropped_queue_frames=dropped_queue_frames,
        dropped_client_buffer_frames=dropped_client_buffer_frames,
        **_metadata_fields(metadata),
    )
    _emit_timeline(
        timeline,
        "entire_call_recording_upload_enqueued",
        session_id=session_id,
        object_name=object_name,
        bytes=bytes,
        audio_duration_ms=duration_ms,
        dropped_queue_frames=dropped_queue_frames,
        dropped_client_buffer_frames=dropped_client_buffer_frames,
        **_metadata_fields(metadata),
    )
    return object_name


class EntireCallRecorder:
    def __init__(
        self,
        *,
        session_id: str,
        sample_rate: int,
        channels: int,
        sample_width: int,
        frame_ms: int,
        bytes_per_frame: int,
        tmp_dir: Path,
        logger_override: Any | None = None,
        timeline: Any | None = None,
        metadata: Mapping[str, Any] | None = None,
        queue_size: int = _DEFAULT_QUEUE_SIZE,
        max_client_buffer_frames: int = _DEFAULT_MAX_CLIENT_BUFFER_FRAMES,
        object_name: str | None = None,
    ) -> None:
        self.session_id = str(session_id or "").strip() or "unknown-session"
        self.sample_rate = int(sample_rate)
        self.source_channels = int(channels)
        if self.source_channels != 1:
            raise ValueError("entire call recorder requires mono source frames")
        self.channels = 2
        self.sample_width = int(sample_width)
        self.frame_ms = int(frame_ms)
        self.bytes_per_frame = int(bytes_per_frame)
        self.object_name = object_name or build_entire_call_object_name(session_id=self.session_id)
        self._tmp_dir = tmp_dir
        self._logger = logger_override or logger
        self._timeline = timeline
        self._metadata = dict(metadata or {})
        self._queue: queue.Queue[_RecordingEvent | object] = queue.Queue(
            maxsize=max(1, int(queue_size))
        )
        self._max_client_buffer_frames = max(1, int(max_client_buffer_frames))
        self._thread: threading.Thread | None = None
        self._state_lock = threading.Lock()
        self._started = False
        self._closed = False
        self._finalized = False
        self._thread_error: BaseException | None = None
        self._frames_written = 0
        self._dropped_queue_frames = 0
        self._dropped_client_buffer_frames = 0
        self._path = self._tmp_dir / (
            f"{_safe_path_part(self.session_id, fallback='unknown-session')}-{uuid.uuid4().hex}.wav"
        )

    @property
    def path(self) -> Path:
        return self._path

    @property
    def dropped_queue_frames(self) -> int:
        return self._dropped_queue_frames

    def start(self) -> None:
        with self._state_lock:
            if self._started:
                return
            self._tmp_dir.mkdir(parents=True, exist_ok=True)
            self._started = True
            self._thread = threading.Thread(
                target=self._run,
                name=f"entire-call-recorder-{self.session_id[:16]}",
                daemon=True,
            )
            self._thread.start()

        log_flow_event(
            self._logger,
            "entire_call_recording_started",
            session_id=self.session_id,
            object_name=self.object_name,
            path=str(self._path),
            sample_rate=self.sample_rate,
            channels=self.channels,
            channel_left="client",
            channel_right="agent",
            frame_ms=self.frame_ms,
            **_metadata_fields(self._metadata),
        )
        _emit_timeline(
            self._timeline,
            "entire_call_recording_started",
            session_id=self.session_id,
            object_name=self.object_name,
            path=str(self._path),
            sample_rate=self.sample_rate,
            channels=self.channels,
            channel_left="client",
            channel_right="agent",
            frame_ms=self.frame_ms,
            **_metadata_fields(self._metadata),
        )

    def record_client_frame(self, frame: bytes) -> None:
        self._enqueue("client", frame)

    def record_output_frame(self, frame: bytes) -> None:
        self._enqueue("output", frame)

    def _enqueue(self, source: str, frame: bytes) -> None:
        try:
            with self._state_lock:
                if not self._started or self._closed:
                    return
            self._queue.put_nowait(_RecordingEvent(source=source, frame=self._normalize_frame(frame)))
        except queue.Full:
            self._record_queue_drop(source=source)
        except Exception:
            self._logger.debug(
                "[entire-call-recording] failed enqueueing frame source=%s",
                source,
                exc_info=True,
            )

    def _record_queue_drop(self, *, source: str) -> None:
        with self._state_lock:
            self._dropped_queue_frames += 1
            dropped = self._dropped_queue_frames

        if dropped == 1 or dropped % 100 == 0:
            log_flow_event(
                self._logger,
                "entire_call_recording_drop",
                session_id=self.session_id,
                object_name=self.object_name,
                source=source,
                reason="queue_full",
                dropped_queue_frames=dropped,
                queue_size=self._queue.qsize(),
                **_metadata_fields(self._metadata),
            )
            _emit_timeline(
                self._timeline,
                "entire_call_recording_drop",
                session_id=self.session_id,
                object_name=self.object_name,
                source=source,
                reason="queue_full",
                dropped_queue_frames=dropped,
                queue_size=self._queue.qsize(),
                **_metadata_fields(self._metadata),
            )

    def _normalize_frame(self, frame: bytes) -> bytes:
        data = bytes(frame or b"")
        if len(data) == self.bytes_per_frame:
            return data
        if len(data) > self.bytes_per_frame:
            return data[: self.bytes_per_frame]
        return data + (b"\x00" * (self.bytes_per_frame - len(data)))

    def _interleave_stereo_frames(self, client_frame: bytes, output_frame: bytes) -> bytes:
        """Build stereo PCM with the client on the left and the agent on the right."""
        stereo = bytearray(len(client_frame) + len(output_frame))
        write_offset = 0
        for sample_offset in range(0, self.bytes_per_frame, self.sample_width):
            next_offset = sample_offset + self.sample_width
            stereo[write_offset : write_offset + self.sample_width] = client_frame[
                sample_offset:next_offset
            ]
            write_offset += self.sample_width
            stereo[write_offset : write_offset + self.sample_width] = output_frame[
                sample_offset:next_offset
            ]
            write_offset += self.sample_width
        return bytes(stereo)

    def _write_stereo_frame(
        self,
        wav_handle: wave.Wave_write,
        *,
        client_frame: bytes,
        output_frame: bytes,
    ) -> None:
        wav_handle.writeframesraw(
            self._interleave_stereo_frames(client_frame, output_frame)
        )
        self._frames_written += 1

    def _run(self) -> None:
        silence = b"\x00" * self.bytes_per_frame
        client_frames: deque[bytes] = deque()
        try:
            with wave.open(str(self._path), "wb") as wav_handle:
                wav_handle.setnchannels(self.channels)
                wav_handle.setsampwidth(self.sample_width)
                wav_handle.setframerate(self.sample_rate)

                while True:
                    event = self._queue.get()
                    try:
                        if event is _SENTINEL:
                            break

                        if not isinstance(event, _RecordingEvent):
                            continue

                        if event.source == "client":
                            if len(client_frames) >= self._max_client_buffer_frames:
                                client_frames.popleft()
                                self._dropped_client_buffer_frames += 1
                            client_frames.append(event.frame)
                            continue

                        if event.source == "output":
                            client_frame = client_frames.popleft() if client_frames else silence
                            self._write_stereo_frame(
                                wav_handle,
                                client_frame=client_frame,
                                output_frame=event.frame,
                            )
                    finally:
                        self._queue.task_done()

                while client_frames:
                    self._write_stereo_frame(
                        wav_handle,
                        client_frame=client_frames.popleft(),
                        output_frame=silence,
                    )
        except BaseException as exc:
            self._thread_error = exc
            self._logger.exception(
                "[entire-call-recording][writer_failed] session_id=%s object_name=%s",
                self.session_id,
                self.object_name,
            )

    def _close_and_join(self) -> None:
        with self._state_lock:
            if not self._started or self._closed:
                return
            self._closed = True

        self._queue.put(_SENTINEL, timeout=5.0)
        thread = self._thread
        if thread is not None:
            thread.join(timeout=5.0)
            if thread.is_alive():
                raise TimeoutError("entire call recorder writer did not stop in time")

    async def finalize(self, *, enqueue_upload: bool = True) -> EntireCallRecordingResult | None:
        with self._state_lock:
            if self._finalized:
                return None
            self._finalized = True

        try:
            await asyncio.to_thread(self._close_and_join)
        except Exception:
            log_flow_event(
                self._logger,
                "entire_call_recording_failed",
                session_id=self.session_id,
                object_name=self.object_name,
                reason="writer_finalize_failed",
                **_metadata_fields(self._metadata),
            )
            _emit_timeline(
                self._timeline,
                "entire_call_recording_failed",
                session_id=self.session_id,
                object_name=self.object_name,
                reason="writer_finalize_failed",
                **_metadata_fields(self._metadata),
            )
            self._logger.exception(
                "[entire-call-recording][finalize_failed] session_id=%s object_name=%s",
                self.session_id,
                self.object_name,
            )
            _cleanup_path(self._path, self._logger)
            return None

        if self._thread_error is not None:
            log_flow_event(
                self._logger,
                "entire_call_recording_failed",
                session_id=self.session_id,
                object_name=self.object_name,
                reason="writer_failed",
                error=type(self._thread_error).__name__,
                **_metadata_fields(self._metadata),
            )
            _emit_timeline(
                self._timeline,
                "entire_call_recording_failed",
                session_id=self.session_id,
                object_name=self.object_name,
                reason="writer_failed",
                error=type(self._thread_error).__name__,
                **_metadata_fields(self._metadata),
            )
            _cleanup_path(self._path, self._logger)
            return None

        file_bytes = self._path.stat().st_size if self._path.exists() else 0
        duration_ms = round(self._frames_written * self.frame_ms)
        result = EntireCallRecordingResult(
            object_name=self.object_name,
            path=self._path,
            bytes=file_bytes,
            frames=self._frames_written,
            duration_ms=duration_ms,
            dropped_queue_frames=self._dropped_queue_frames,
            dropped_client_buffer_frames=self._dropped_client_buffer_frames,
            upload_enqueued=False,
        )
        log_flow_event(
            self._logger,
            "entire_call_recording_finalized",
            session_id=self.session_id,
            object_name=self.object_name,
            path=str(self._path),
            bytes=file_bytes,
            frames=self._frames_written,
            audio_duration_ms=duration_ms,
            channels=self.channels,
            channel_left="client",
            channel_right="agent",
            dropped_queue_frames=self._dropped_queue_frames,
            dropped_client_buffer_frames=self._dropped_client_buffer_frames,
            **_metadata_fields(self._metadata),
        )
        _emit_timeline(
            self._timeline,
            "entire_call_recording_finalized",
            session_id=self.session_id,
            object_name=self.object_name,
            path=str(self._path),
            bytes=file_bytes,
            frames=self._frames_written,
            audio_duration_ms=duration_ms,
            channels=self.channels,
            channel_left="client",
            channel_right="agent",
            dropped_queue_frames=self._dropped_queue_frames,
            dropped_client_buffer_frames=self._dropped_client_buffer_frames,
            **_metadata_fields(self._metadata),
        )

        if not enqueue_upload:
            return result

        queued_object = enqueue_entire_call_upload(
            path=self._path,
            object_name=self.object_name,
            session_id=self.session_id,
            bytes=file_bytes,
            duration_ms=duration_ms,
            dropped_queue_frames=self._dropped_queue_frames,
            dropped_client_buffer_frames=self._dropped_client_buffer_frames,
            metadata=self._metadata,
            logger_override=self._logger,
            timeline=self._timeline,
        )
        return EntireCallRecordingResult(
            object_name=self.object_name,
            path=self._path,
            bytes=file_bytes,
            frames=self._frames_written,
            duration_ms=duration_ms,
            dropped_queue_frames=self._dropped_queue_frames,
            dropped_client_buffer_frames=self._dropped_client_buffer_frames,
            upload_enqueued=queued_object is not None,
        )


def create_entire_call_recorder_from_env(
    *,
    session_id: str,
    sample_rate: int,
    channels: int,
    sample_width: int,
    frame_ms: int,
    bytes_per_frame: int,
    logger_override: Any | None = None,
    timeline: Any | None = None,
    metadata: Mapping[str, Any] | None = None,
) -> EntireCallRecorder | None:
    active_logger = logger_override or logger
    metadata = metadata or {}
    if not _env_bool("ENTIRE_CALL_RECORDING_ENABLED", True):
        log_flow_event(
            active_logger,
            "entire_call_recording_disabled",
            session_id=session_id,
            reason="disabled_by_env",
            **_metadata_fields(metadata),
        )
        _emit_timeline(
            timeline,
            "entire_call_recording_disabled",
            session_id=session_id,
            reason="disabled_by_env",
            **_metadata_fields(metadata),
        )
        return None

    if _oci_upload_config_from_env() is None:
        log_flow_event(
            active_logger,
            "entire_call_recording_disabled",
            session_id=session_id,
            reason="missing_oci_config",
            **_metadata_fields(metadata),
        )
        _emit_timeline(
            timeline,
            "entire_call_recording_disabled",
            session_id=session_id,
            reason="missing_oci_config",
            **_metadata_fields(metadata),
        )
        return None

    recorder = EntireCallRecorder(
        session_id=session_id,
        sample_rate=sample_rate,
        channels=channels,
        sample_width=sample_width,
        frame_ms=frame_ms,
        bytes_per_frame=bytes_per_frame,
        tmp_dir=_tmp_dir_from_env(),
        logger_override=active_logger,
        timeline=timeline,
        metadata=metadata,
    )
    try:
        recorder.start()
    except Exception:
        log_flow_event(
            active_logger,
            "entire_call_recording_disabled",
            session_id=session_id,
            reason="start_failed",
            **_metadata_fields(metadata),
        )
        _emit_timeline(
            timeline,
            "entire_call_recording_disabled",
            session_id=session_id,
            reason="start_failed",
            **_metadata_fields(metadata),
        )
        active_logger.exception(
            "[entire-call-recording][start_failed] session_id=%s",
            session_id,
        )
        return None
    return recorder
