queue_processor.py

import os
import gc
import ctypes
import asyncio
import time
import logging
import glob

from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from telegram.error import RetryAfter, TelegramError

from config import load_config, check_disk_space, check_ffmpeg, DOWNLOAD_DIR, get_ffmpeg_command, get_proxy_list, get_cookie_file
from downloader import download_content, get_video_info, get_playlist_info
from uploader import upload_video_streaming, upload_audio_streaming, split_video, crop_to_square
from handlers import cancelled_tasks, stopped_tasks, fromstart_tasks

logger = logging.getLogger(__name__)

def _free_memory():
    """Force garbage collection and release memory back to OS via glibc malloc_trim."""
    gc.collect()
    try:
        ctypes.CDLL("libc.so.6").malloc_trim(0)
    except Exception:
        pass

def _cleanup_partial_downloads():
    """Remove .part files and orphaned thumbnails from downloads directory."""
    for f in glob.glob(os.path.join(DOWNLOAD_DIR, "*.part")):
        try: os.remove(f)
        except: pass
    for ext in ('*.jpg', '*.webp', '*.jpeg'):
        for f in glob.glob(os.path.join(DOWNLOAD_DIR, ext)):
            base = os.path.splitext(f)[0]
            has_video = any(os.path.exists(base + v) for v in ('.mp4', '.mkv', '.webm', '.m4a', '.mp3'))
            if not has_video:
                try: os.remove(f)
                except: pass

async def tg_retry(func, *args, **kwargs):
    """Retry Telegram API calls up to 10 times on RateLimit."""
    max_retries = 10
    for attempt in range(max_retries):
        try:
            return await func(*args, **kwargs)
        except RetryAfter as e:
            wait_time = e.retry_after
            logger.warning(f"Flood control: Waiting {wait_time}s (Attempt {attempt+1}/10)")
            await asyncio.sleep(wait_time)
        except TelegramError as e:
            if "Flood control" in str(e):
                logger.warning(f"Flood caught via error msg: {e} (Attempt {attempt+1}/10)")
                await asyncio.sleep(5)
                continue
            raise e
        except Exception as e:
            if attempt == max_retries - 1:
                raise e
            logger.warning(f"Unexpected error in tg_retry: {e}. Retrying...")
            await asyncio.sleep(2)
    raise Exception("Max retries exceeded for Telegram API call")

async def handle_upload(application, chat_id, file_path, title, url, audio_only=False, update_status_func=None, channel_name=None, reply_to_message_id=None, thumb_path=None):
    """Helper to handle video/audio upload with splitting and cleanup."""
    try:
        if audio_only:
            if update_status_func:
                await update_status_func("⬆️ Uploading audio...", force=True)
            
            config = load_config()
            api_url = config.get('api_url', '')
            bot_token = config.get('bot_token', '')
            is_local_api = api_url and 'api.telegram.org' not in api_url
            
            if channel_name:
                full_caption = f"{channel_name}\n{title}\n{url}"
            else:
                full_caption = f"{title}\n{url}"
            if is_local_api:
                await upload_audio_streaming(bot_token, api_url, chat_id, file_path, title, full_caption, reply_to_message_id=reply_to_message_id, thumb_path=thumb_path)
            else:
                with open(file_path, 'rb') as f:
                    if thumb_path and os.path.exists(thumb_path):
                        thumb_path = crop_to_square(thumb_path)
                        thumb = open(thumb_path, 'rb')
                    else:
                        thumb = None
                    await tg_retry(application.bot.send_audio, chat_id=chat_id, audio=f, title=title, caption=full_caption, reply_to_message_id=reply_to_message_id, thumbnail=thumb)
                    if thumb: thumb.close()
            
            if os.path.exists(file_path):
                os.remove(file_path)
        
        else:
            # Video upload with splitting
            if update_status_func:
                await update_status_func("✂️ Checking file size...", force=True)
            
            loop = asyncio.get_running_loop()
            if not check_ffmpeg():
                files_to_upload = [file_path]
            else:
                files_to_upload = await loop.run_in_executor(None, split_video, file_path)
            
            total_parts = len(files_to_upload)
            
            for i, f_path in enumerate(files_to_upload):
                if channel_name:
                    caption = f"{channel_name}\n{title}\n{url}"
                else:
                    caption = f"{title}\n{url}"
                    
                if total_parts > 1:
                    if channel_name:
                        caption = f"{channel_name}\n{title} (Part {i+1}/{total_parts})\n{url}"
                    else:
                        caption = f"{title} (Part {i+1}/{total_parts})\n{url}"
                
                if update_status_func:
                    await update_status_func(f"⬆️ Uploading part {i+1}/{total_parts}...", force=True)
                
                try:
                    config = load_config()
                    api_url = config.get('api_url', '')
                    bot_token = config.get('bot_token', '')
                    is_local_api = api_url and 'api.telegram.org' not in api_url

                    # Only add audio button if callback_data fits Telegram's 64-byte limit
                    audio_cb_data = f"audio:{url}"
                    if len(audio_cb_data.encode('utf-8')) <= 64:
                        reply_markup_dict = {"inline_keyboard": [[{"text": "🎵 Download Audio", "callback_data": audio_cb_data}]]}
                        reply_markup = InlineKeyboardMarkup([[InlineKeyboardButton("🎵 Download Audio", callback_data=audio_cb_data)]])
                    else:
                        reply_markup_dict = None
                        reply_markup = None

                    if is_local_api:
                        await upload_video_streaming(bot_token, api_url, chat_id, f_path, caption, reply_markup_dict, reply_to_message_id=reply_to_message_id, thumb_path=thumb_path)
                    else:
                        with open(f_path, 'rb') as f:
                            if thumb_path and os.path.exists(thumb_path):
                                thumb_path = crop_to_square(thumb_path)
                                thumb = open(thumb_path, 'rb')
                            else:
                                thumb = None

                            await tg_retry(application.bot.send_video,
                                chat_id=chat_id, video=f, caption=caption,
                                supports_streaming=True, reply_markup=reply_markup,
                                reply_to_message_id=reply_to_message_id,
                                thumbnail=thumb
                            )
                            if thumb: thumb.close()
                except Exception as e:
                    logger.error(f"Upload failed for part {i+1}: {e}")
                    await tg_retry(application.bot.send_message, chat_id=chat_id, text=f"❌ Upload failed for part {i+1}: {e}")
            
            # Cleanup
            if thumb_path and os.path.exists(thumb_path):
                os.remove(thumb_path)
                
            if update_status_func:
                await update_status_func("🧹 Cleaning up...", force=True)
            
            if os.path.exists(file_path):
                os.remove(file_path)
            for f_path in files_to_upload:
                if os.path.exists(f_path) and f_path != file_path:
                    os.remove(f_path)
                    
    except Exception as e:
        logger.error(f"Error in handle_upload: {e}")
        error_text = f"🔥 Upload error: {e}"
        if update_status_func:
            await update_status_func(error_text, force=True)
        else:
            await application.bot.send_message(chat_id=chat_id, text=error_text)
        if file_path and os.path.exists(file_path):
            try: os.remove(file_path)
            except: pass
        if thumb_path and os.path.exists(thumb_path):
            try: os.remove(thumb_path)
            except: pass
    finally:
        _free_memory()

async def process_queue(application, request_queue):
    """Main queue processor for single video downloads."""
    logger.info("Queue processor started.")
    
    while True:
        task = await request_queue.get()
        try:
            status_msg_passed = None
            is_live = False
            channel_name = None
            if len(task) == 7:
                chat_id, url, message_id, max_height, status_msg_passed, channel_name, is_live = task
            elif len(task) == 6:
                chat_id, url, message_id, max_height, status_msg_passed, channel_name = task
            elif len(task) == 5:
                chat_id, url, message_id, max_height, status_msg_passed = task
            elif len(task) == 4:
                chat_id, url, message_id, max_height = task
            else:
                chat_id, url, message_id = task
                max_height = 1080
            
            audio_only = (max_height in (-1, -2))
            audio_format = 'mp3' if max_height == -2 else 'm4a'
            if audio_only:
                max_height = 1080
            
            task_id = f"{chat_id}_{message_id}_{int(time.time())}"
            status_msg = status_msg_passed
            last_edit_time = 0
            
            async def update_status_msg(text, force=False, show_cancel=False):
                nonlocal status_msg, last_edit_time
                now = time.time()
                if not force and (now - last_edit_time < 20):
                    return
                try:
                    reply_markup = None
                    if show_cancel:
                        keyboard = [[InlineKeyboardButton("❌ Cancel", callback_data=f"cancel:{task_id}")]]
                        reply_markup = InlineKeyboardMarkup(keyboard)
                    if status_msg:
                        if status_msg.text != text:
                            await tg_retry(status_msg.edit_text, text, reply_markup=reply_markup)
                            last_edit_time = now
                    else:
                        status_msg = await tg_retry(application.bot.send_message,
                            chat_id=chat_id, text=text, reply_to_message_id=message_id, reply_markup=reply_markup
                        )
                        last_edit_time = now
                except Exception as e:
                    logger.warning(f"Failed to update status: {e}")

            # Initial Live Detection (from queue flag)
            if is_live:
                asyncio.create_task(process_live_stream(application, chat_id, url, message_id, status_msg, task_id, update_status_msg, channel_name))
                continue
                
            await update_status_msg(f"🚀 Processing: {url}", force=True, show_cancel=True)
            
            # Info extraction and secondary Live Detection
            await update_status_msg("📊 Checking video info...", force=True, show_cancel=True)
            video_info = {}
            try:
                loop = asyncio.get_running_loop()
                # 45s timeout for extraction to avoid blocking the queue permanently
                video_info = await asyncio.wait_for(
                    loop.run_in_executor(None, lambda: get_video_info(url)),
                    timeout=45
                )
                
                # If info extraction reveals it IS a live stream, handle it
                if video_info.get('is_live'):
                    logger.info(f"URL detected as LIVE during info check: {url}")
                    channel_name = channel_name or video_info.get('uploader') or video_info.get('title', 'Live')
                    asyncio.create_task(process_live_stream(application, chat_id, url, message_id, status_msg, task_id, update_status_msg, channel_name))
                    continue
            except asyncio.TimeoutError:
                logger.warning(f"Timeout checking info for {url}, proceeding with defaults")
            except Exception as e:
                logger.error(f"Error checking video info: {e}")
            
            # Disk space check
            config = load_config()
            max_disk_gb = config.get('max_disk_gb', 0)
            if max_disk_gb > 0:
                estimated_mb = video_info.get('filesize_mb', 0)
                if estimated_mb > 0:
                    can_download, remaining_gb = check_disk_space(estimated_mb)
                    if not can_download:
                        await update_status_msg(f"❌ Low disk space! Need {estimated_mb/1024:.1f}GB, have {remaining_gb:.1f}GB.", force=True)
                        continue

            if task_id in cancelled_tasks:
                if status_msg:
                    try: await tg_retry(status_msg.delete)
                    except: pass
                cancelled_tasks.discard(task_id)
                continue

            loop = asyncio.get_running_loop()
            def progress_cb(d):
                if task_id in cancelled_tasks: raise Exception("Download cancelled")
                if d['status'] == 'downloading':
                    p = d.get('_percent_str', '0%')
                    eta = d.get('_eta_str', '?')
                    mode = f"🎵 Audio {audio_format.upper()}" if audio_only else f"{max_height}p"
                    asyncio.run_coroutine_threadsafe(update_status_msg(f"⬇️ Downloading ({mode}): {p}\nETA: {eta}", show_cancel=True), loop)

            # Download
            try:
                file_path, title, video_id, thumb_path = await loop.run_in_executor(
                    None, 
                    lambda: download_content(url, progress_cb, audio_only=audio_only, audio_format=audio_format, max_height=max_height, task_id=task_id, cancelled_tasks=cancelled_tasks)
                )
                # Upload using helper
                await handle_upload(application, chat_id, file_path, title, url, audio_only, update_status_msg, channel_name, message_id, thumb_path)
            except Exception as e:
                # Cleanup potential partial files on failure
                logger.error(f"Download failed for {url}: {e}")
                _cleanup_partial_downloads()

                await update_status_msg(f"❌ Download failed: {e}", force=True)
                continue
            
            # Delete the progress/status message upon completion
            if status_msg:
                try:
                    await tg_retry(status_msg.delete)
                except Exception as e:
                    logger.warning(f"Failed to delete status message: {e}")

        except Exception as e:
            logger.error(f"Error in process_queue: {e}")
            await update_status_msg(f"🔥 Error: {e}", force=True)
        finally:
            request_queue.task_done()
            _free_memory()

async def _kill_process(process, task_id):
    """Gracefully stop process: SIGINT → SIGTERM → SIGKILL."""
    import signal
    try:
        process.send_signal(signal.SIGINT)
        await asyncio.wait_for(process.wait(), timeout=20)
        logger.info(f"[LIVE:{task_id}] Process stopped gracefully via SIGINT")
    except asyncio.TimeoutError:
        logger.warning(f"[LIVE:{task_id}] SIGINT timeout, sending SIGTERM")
        try:
            process.terminate()
            await asyncio.wait_for(process.wait(), timeout=15)
            logger.info(f"[LIVE:{task_id}] Process terminated via SIGTERM")
        except asyncio.TimeoutError:
            logger.warning(f"[LIVE:{task_id}] SIGTERM timeout, sending SIGKILL")
            try:
                process.kill()
                await asyncio.wait_for(process.wait(), timeout=10)
            except Exception:
                logger.error(f"[LIVE:{task_id}] SIGKILL also failed, process may be orphaned")
    except Exception as e:
        logger.error(f"[LIVE:{task_id}] _kill_process error: {e}", exc_info=True)

async def process_live_stream(application, chat_id, url, message_id, status_msg, task_id, update_status_msg, channel_name):
    """Record live stream using yt-dlp with forced HLS (iOS client).
    HLS writes continuously via --hls-use-mpegts, solving the DASH fragment problem."""
    SEGMENT_SIZE_BYTES = 1900 * 1024 * 1024  # 1.9GB per segment
    logger.info(f"[LIVE:{task_id}] START url={url}, chat_id={chat_id}, channel={channel_name}")

    fromstart_triggered = False

    def _make_keyboard():
        buttons = []
        if not fromstart_triggered:
            buttons.append(InlineKeyboardButton("⏪ From Start", callback_data=f"fromstart:{task_id}"))
        buttons.append(InlineKeyboardButton("⏹ Stop & Upload", callback_data=f"stoplive:{task_id}"))
        buttons.append(InlineKeyboardButton("❌ Cancel", callback_data=f"cancel:{task_id}"))
        return InlineKeyboardMarkup([buttons])

    async def live_status(text):
        nonlocal status_msg
        try:
            keyboard = _make_keyboard()
            logger.info(f"[LIVE:{task_id}] live_status: '{text}'")
            if status_msg:
                if status_msg.text != text:
                    await tg_retry(status_msg.edit_text, text, reply_markup=keyboard)
            else:
                status_msg = await tg_retry(
                    application.bot.send_message,
                    chat_id=chat_id, text=text, reply_to_message_id=message_id, reply_markup=keyboard
                )
        except Exception as e:
            logger.error(f"[LIVE:{task_id}] live_status failed: {e}", exc_info=True)

    def _build_record_cmd(output_path, proxy=None):
        """streamlink command for live recording — writes continuously to file."""
        cmd = [
            'streamlink',
            '--force',
            '--loglevel', 'warning',
            '--ffmpeg-ffmpeg', get_ffmpeg_command(),
            '-o', output_path,
        ]
        if proxy:
            cmd += ['--http-proxy', proxy]
        cmd += [url, 'best']
        return cmd

    def _build_fromstart_cmd(proxy=None):
        """yt-dlp command for from-start download, outputs to stdout."""
        cmd = [
            'yt-dlp',
            '--no-part',
            '--no-check-certificates',
            '--no-playlist',
            '--hls-use-mpegts',
            '--live-from-start',
            '--ffmpeg-location', get_ffmpeg_command(),
            '--socket-timeout', '30',
            '--retries', '10',
            '--fragment-retries', '10',
            '-o', '-',
        ]
        cookie_file = get_cookie_file()
        if cookie_file:
            cmd += ['--cookies', cookie_file]
        if proxy:
            cmd += ['--proxy', proxy]
        cmd.append(url)
        return cmd

    async def _download_from_start():
        """Background task: download the stream from beginning using yt-dlp via pipe.
        Reads stdout directly into 1.9GB segment files — no giant .ts accumulating on disk."""
        bg_id = f"{task_id}_fromstart"
        logger.info(f"[LIVE:{bg_id}] Background from-start download starting (pipe mode)")
        proxy_list = get_proxy_list()

        proc = None
        for proxy in proxy_list:
            cmd = _build_fromstart_cmd(proxy)
            logger.info(f"[LIVE:{bg_id}] cmd: {' '.join(cmd[:8])}...")
            try:
                proc = await asyncio.create_subprocess_exec(
                    *cmd,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.DEVNULL,
                )
                logger.info(f"[LIVE:{bg_id}] pid={proc.pid}")
                break
            except Exception as e:
                logger.error(f"[LIVE:{bg_id}] Spawn failed: {e}", exc_info=True)
                continue

        if proc is None:
            logger.error(f"[LIVE:{bg_id}] All proxies failed to spawn")
            try:
                await tg_retry(
                    application.bot.send_message,
                    chat_id=chat_id,
                    text="❌ From-start download failed.",
                    reply_to_message_id=message_id
                )
            except Exception:
                pass
            return

        seg_num = 0
        seg_file = None
        seg_path = None
        seg_bytes = 0
        total_bytes = 0
        start_time = time.time()
        got_data = False

        try:
            while True:
                # Check signals
                if task_id in cancelled_tasks:
                    await _kill_process(proc, bg_id)
                    _cleanup_live_files(bg_id)
                    logger.info(f"[LIVE:{bg_id}] Cancelled")
                    if seg_file:
                        seg_file.close()
                        try: os.remove(seg_path)
                        except: pass
                    return
                if task_id in stopped_tasks:
                    logger.info(f"[LIVE:{bg_id}] Stop signal received")
                    await _kill_process(proc, bg_id)
                    break

                # Read chunk from pipe (non-blocking with timeout)
                try:
                    chunk = await asyncio.wait_for(proc.stdout.read(1024 * 1024), timeout=5)
                except asyncio.TimeoutError:
                    # No data yet — check if process died
                    if proc.returncode is not None:
                        break
                    # VOD unavailable check
                    elapsed = time.time() - start_time
                    if elapsed > 90 and not got_data:
                        logger.warning(f"[LIVE:{bg_id}] No data after {elapsed:.0f}s, VOD likely unavailable")
                        await _kill_process(proc, bg_id)
                        try:
                            await tg_retry(
                                application.bot.send_message,
                                chat_id=chat_id,
                                text="❌ From-start download failed (VOD unavailable).",
                                reply_to_message_id=message_id
                            )
                        except Exception:
                            pass
                        _cleanup_live_files(bg_id)
                        return
                    continue

                if not chunk:
                    # EOF — yt-dlp finished
                    break

                got_data = True
                total_bytes += len(chunk)

                # Open new segment file if needed
                if seg_file is None:
                    seg_num += 1
                    seg_path = os.path.join(DOWNLOAD_DIR, f"live_{bg_id}_seg{seg_num:03d}.ts")
                    seg_file = open(seg_path, 'wb')
                    seg_bytes = 0

                seg_file.write(chunk)
                seg_bytes += len(chunk)

                # Segment full — close, remux+upload, start next
                if seg_bytes >= SEGMENT_SIZE_BYTES:
                    seg_file.close()
                    seg_file = None
                    seg_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{bg_id}_seg{seg_num:03d}.mp4")
                    logger.info(f"[LIVE:{bg_id}] Segment {seg_num} complete: {seg_bytes/(1024*1024):.1f}MB (total: {total_bytes/(1024*1024):.0f}MB)")
                    asyncio.create_task(_remux_and_upload_bg(bg_id, seg_path, seg_mp4, seg_num))

        except Exception as e:
            logger.error(f"[LIVE:{bg_id}] Pipe read error: {e}", exc_info=True)
        finally:
            # Close last segment and upload if it has data
            if seg_file:
                seg_file.close()
                if seg_bytes > 1024:
                    seg_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{bg_id}_seg{seg_num:03d}.mp4")
                    logger.info(f"[LIVE:{bg_id}] Final segment {seg_num}: {seg_bytes/(1024*1024):.1f}MB (total: {total_bytes/(1024*1024):.0f}MB)")
                    await _remux_and_upload_bg(bg_id, seg_path, seg_mp4, seg_num, is_final=True)
                else:
                    try: os.remove(seg_path)
                    except: pass
            elif seg_path and os.path.exists(seg_path) and os.path.getsize(seg_path) > 1024:
                # Edge case: segment was closed by size limit but we need to mark last uploaded as final
                pass

            # Wait for proc to finish if still running
            if proc.returncode is None:
                try:
                    await asyncio.wait_for(proc.wait(), timeout=5)
                except asyncio.TimeoutError:
                    await _kill_process(proc, bg_id)

            _cleanup_live_files(bg_id)
            logger.info(f"[LIVE:{bg_id}] Complete. Segments: {seg_num}, Total: {total_bytes/(1024*1024):.1f}MB")

    async def _remux_and_upload_bg(bg_id, ts_path, mp4_path, seg_num, is_final=False):
        """Remux a from-start segment and upload."""
        try:
            ts_size = os.path.getsize(ts_path) if os.path.exists(ts_path) else 0
            if ts_size == 0:
                try: os.remove(ts_path)
                except: pass
                return
            logger.info(f"[LIVE:{bg_id}] Remuxing seg {seg_num}: {ts_size/(1024*1024):.1f}MB")
            remux = await asyncio.create_subprocess_exec(
                get_ffmpeg_command(), '-y',
                '-err_detect', 'ignore_err',
                '-fflags', '+genpts+discardcorrupt',
                '-i', ts_path,
                '-c', 'copy', '-movflags', '+faststart', mp4_path,
                stdout=asyncio.subprocess.DEVNULL,
                stderr=asyncio.subprocess.PIPE,
            )
            await remux.stderr.read()
            await remux.wait()
            if remux.returncode != 0:
                remux2 = await asyncio.create_subprocess_exec(
                    get_ffmpeg_command(), '-y',
                    '-err_detect', 'ignore_err',
                    '-fflags', '+genpts+discardcorrupt',
                    '-i', ts_path,
                    '-c', 'copy', mp4_path,
                    stdout=asyncio.subprocess.DEVNULL,
                    stderr=asyncio.subprocess.DEVNULL,
                )
                await remux2.wait()
            try: os.remove(ts_path)
            except: pass

            if os.path.exists(mp4_path) and os.path.getsize(mp4_path) > 0:
                title = f"⏪ {channel_name} - From Start Part {seg_num}"
                if is_final:
                    title += " (End)"
                logger.info(f"[LIVE:{bg_id}] Uploading seg {seg_num}: {os.path.getsize(mp4_path)/(1024*1024):.1f}MB")
                await handle_upload(application, chat_id, mp4_path, title, url, False, None, channel_name, message_id)
                logger.info(f"[LIVE:{bg_id}] Upload done seg {seg_num}")
            else:
                logger.error(f"[LIVE:{bg_id}] Remux produced no output for seg {seg_num}")
        except Exception as e:
            logger.error(f"[LIVE:{bg_id}] Remux/upload seg {seg_num} error: {e}", exc_info=True)

    async def _concat_parts(part_files, output_ts):
        """Concatenate multiple .ts part files using binary concat (TS is designed for this)."""
        if len(part_files) == 1:
            os.rename(part_files[0], output_ts)
            return True
        try:
            with open(output_ts, 'wb') as out:
                for p in part_files:
                    if os.path.exists(p) and os.path.getsize(p) > 0:
                        with open(p, 'rb') as inp:
                            while True:
                                chunk = inp.read(8 * 1024 * 1024)
                                if not chunk:
                                    break
                                out.write(chunk)
            if os.path.exists(output_ts) and os.path.getsize(output_ts) > 0:
                for p in part_files:
                    try: os.remove(p)
                    except: pass
                return True
            else:
                logger.error(f"[LIVE:{task_id}] binary concat produced empty file")
                try: os.remove(output_ts)
                except: pass
                return False
        except Exception as e:
            logger.error(f"[LIVE:{task_id}] binary concat error: {e}")
            try: os.remove(output_ts)
            except: pass
            return False

    async def _remux_and_upload(ts_path, mp4_path, seg_num, is_final=False):
        """Background task: remux .ts to .mp4 and upload."""
        try:
            ts_size = os.path.getsize(ts_path) if os.path.exists(ts_path) else 0
            if ts_size == 0:
                try: os.remove(ts_path)
                except: pass
                return

            logger.info(f"[LIVE:{task_id}] BG remux seg {seg_num}: {ts_size/(1024*1024):.1f}MB")
            remux = await asyncio.create_subprocess_exec(
                get_ffmpeg_command(), '-y',
                '-err_detect', 'ignore_err',
                '-fflags', '+genpts+discardcorrupt',
                '-i', ts_path,
                '-c', 'copy', '-movflags', '+faststart', mp4_path,
                stdout=asyncio.subprocess.DEVNULL,
                stderr=asyncio.subprocess.PIPE,
            )
            await remux.stderr.read()
            await remux.wait()
            if remux.returncode != 0:
                remux2 = await asyncio.create_subprocess_exec(
                    get_ffmpeg_command(), '-y',
                    '-err_detect', 'ignore_err',
                    '-fflags', '+genpts+discardcorrupt',
                    '-i', ts_path,
                    '-c', 'copy', mp4_path,
                    stdout=asyncio.subprocess.DEVNULL,
                    stderr=asyncio.subprocess.PIPE,
                )
                await remux2.stderr.read()
                await remux2.wait()
                if remux2.returncode != 0:
                    logger.error(f"[LIVE:{task_id}] BG remux seg {seg_num} failed completely")
                    try: os.remove(ts_path)
                    except: pass
                    return
            try: os.remove(ts_path)
            except: pass

            upload_size = os.path.getsize(mp4_path) if os.path.exists(mp4_path) else 0
            if upload_size > 0:
                title = f"\U0001f534 {channel_name} - LIVE Part {seg_num}"
                if is_final:
                    title += " (End)"
                logger.info(f"[LIVE:{task_id}] BG uploading seg {seg_num}: {upload_size/(1024*1024):.1f}MB")
                await handle_upload(application, chat_id, mp4_path, title, url, False, None, channel_name, message_id)
                logger.info(f"[LIVE:{task_id}] BG upload done seg {seg_num}")
            else:
                logger.warning(f"[LIVE:{task_id}] BG remux produced empty file seg {seg_num}")
        except Exception as e:
            logger.error(f"[LIVE:{task_id}] BG remux/upload seg {seg_num} error: {e}", exc_info=True)

    async def _start_recording(part_path, proxy_list):
        """Start a streamlink recording process, trying each proxy.
        Waits a few seconds to verify the process doesn't die immediately.
        Returns (process, proxy) or (None, None)."""
        for proxy in proxy_list:
            cmd = _build_record_cmd(part_path, proxy)
            logger.info(f"[LIVE:{task_id}] streamlink cmd (proxy={proxy}): {' '.join(cmd[:10])}...")
            try:
                proc = await asyncio.create_subprocess_exec(
                    *cmd,
                    stdout=asyncio.subprocess.DEVNULL,
                    stderr=asyncio.subprocess.PIPE,
                )
                logger.info(f"[LIVE:{task_id}] streamlink pid={proc.pid} proxy={proxy}")
            except Exception as e:
                logger.error(f"[LIVE:{task_id}] Spawn failed proxy={proxy}: {e}")
                continue

            # Wait up to 5s to verify process doesn't die immediately
            for _ in range(5):
                await asyncio.sleep(1)
                if proc.returncode is not None:
                    break
                file_size = os.path.getsize(part_path) if os.path.exists(part_path) else 0
                if file_size > 0:
                    break

            if proc.returncode is not None:
                # Process died — read stderr for diagnosis, try next proxy
                stderr_out = b''
                try:
                    stderr_out = await asyncio.wait_for(proc.stderr.read(), timeout=3)
                except Exception:
                    pass
                logger.warning(f"[LIVE:{task_id}] streamlink died immediately with proxy={proxy} rc={proc.returncode}: {stderr_out.decode(errors='replace')[:200]}")
                if os.path.exists(part_path) and os.path.getsize(part_path) == 0:
                    try: os.remove(part_path)
                    except: pass
                continue

            # Process survived — drain stderr pipe in background to prevent deadlock
            asyncio.create_task(_drain_stderr(proc, proxy))
            return proc, proxy

        return None, None

    async def _drain_stderr(proc, proxy):
        """Continuously drain stderr to prevent pipe buffer deadlock."""
        try:
            while True:
                chunk = await proc.stderr.read(4096)
                if not chunk:
                    break
        except Exception:
            pass

    def _get_total_parts_size(part_files):
        """Get total size of all part files."""
        total = 0
        for p in part_files:
            try:
                total += os.path.getsize(p) if os.path.exists(p) else 0
            except OSError:
                pass
        return total

    try:
        await live_status(f"\U0001f534 Starting live recording: {channel_name}")
        os.makedirs(DOWNLOAD_DIR, exist_ok=True)

        proxy_list = get_proxy_list()
        logger.info(f"[LIVE:{task_id}] Proxies: {proxy_list}")
        segment_num = 0
        uploaded_segments = []
        consecutive_failures = 0
        bg_tasks = []

        # Parts accumulate until size limit, then get concat'd into a segment
        part_num = 0
        part_files = []

        # Start first part
        part_num = 1
        current_part = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_p{part_num:03d}.ts")
        part_files.append(current_part)
        process, used_proxy = await _start_recording(current_part, proxy_list)

        if process is None:
            await update_status_msg("❌ Failed to start recording. All proxies failed.", force=True)
            _cleanup_live_files(task_id)
            return

        await live_status(f"\U0001f534 Recording live stream: {channel_name}")
        poll_count = 0

        while True:
            if process.returncode is not None:
                logger.info(f"[LIVE:{task_id}] streamlink exited rc={process.returncode}")

                if process.returncode != 0:
                    current_size = os.path.getsize(current_part) if os.path.exists(current_part) else 0
                    logger.warning(f"[LIVE:{task_id}] streamlink error rc={process.returncode}, part_size={current_size}")

                    if current_size == 0 and not uploaded_segments and _get_total_parts_size(part_files) == 0:
                        # Nothing recorded at all — might not be live
                        pass

                    if current_size == 0:
                        # Remove empty part from list
                        part_files = [p for p in part_files if p != current_part]
                        try: os.remove(current_part)
                        except: pass

                        consecutive_failures += 1
                        if consecutive_failures >= 3:
                            # Stream probably dead — upload what we have
                            if part_files:
                                segment_num += 1
                                seg_ts = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.ts")
                                seg_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.mp4")
                                if await _concat_parts(part_files, seg_ts):
                                    bg_tasks.append(asyncio.create_task(_remux_and_upload(seg_ts, seg_mp4, segment_num, is_final=True)))
                                    uploaded_segments.append(segment_num)
                                else:
                                    for pi, p in enumerate(part_files, 1):
                                        if os.path.exists(p) and os.path.getsize(p) > 1024:
                                            p_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_fallback{pi:03d}.mp4")
                                            bg_tasks.append(asyncio.create_task(_remux_and_upload(p, p_mp4, pi, is_final=(pi == len(part_files)))))
                                    uploaded_segments.append(segment_num)
                                part_files = []
                            elif not uploaded_segments:
                                await update_status_msg(f"❌ Recording failed after {consecutive_failures} attempts.", force=True)
                            logger.error(f"[LIVE:{task_id}] {consecutive_failures} consecutive failures, giving up")
                            break
                        logger.info(f"[LIVE:{task_id}] No data, retry {consecutive_failures}/3 in 5s")
                        await asyncio.sleep(5)
                        # Reuse same part path
                        current_part = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_p{part_num:03d}.ts")
                        part_files.append(current_part)
                        process, used_proxy = await _start_recording(current_part, proxy_list)
                        if process is None:
                            break
                        continue
                    else:
                        consecutive_failures = 0
                        # Had data but crashed — check if we need to segment first
                        total_size = _get_total_parts_size(part_files)
                        if total_size >= SEGMENT_SIZE_BYTES:
                            logger.info(f"[LIVE:{task_id}] Crash recovery: size {total_size/(1024*1024):.1f}MB >= limit, segmenting")
                            segment_num += 1
                            seg_ts = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.ts")
                            seg_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.mp4")
                            if await _concat_parts(part_files, seg_ts):
                                bg_tasks.append(asyncio.create_task(_remux_and_upload(seg_ts, seg_mp4, segment_num)))
                                uploaded_segments.append(segment_num)
                            else:
                                for pi, p in enumerate(part_files, 1):
                                    if os.path.exists(p) and os.path.getsize(p) > 1024:
                                        p_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_fallback{pi:03d}.mp4")
                                        bg_tasks.append(asyncio.create_task(_remux_and_upload(p, p_mp4, pi)))
                                uploaded_segments.append(segment_num)
                            part_files = []
                        # Start new part
                        part_num += 1
                        current_part = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_p{part_num:03d}.ts")
                        part_files.append(current_part)
                        await live_status(f"\U0001f534 Reconnecting: {channel_name}")
                        await asyncio.sleep(3)
                        process, used_proxy = await _start_recording(current_part, proxy_list)
                        if process is None:
                            break
                        await live_status(f"\U0001f534 Recording live stream: {channel_name}")
                        continue
                else:
                    # rc=0: streamlink exited cleanly — check size before restarting
                    consecutive_failures = 0
                    logger.info(f"[LIVE:{task_id}] streamlink exited rc=0, trying to continue...")
                    total_size = _get_total_parts_size(part_files)
                    if total_size >= SEGMENT_SIZE_BYTES:
                        logger.info(f"[LIVE:{task_id}] Clean exit: size {total_size/(1024*1024):.1f}MB >= limit, segmenting")
                        segment_num += 1
                        seg_ts = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.ts")
                        seg_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.mp4")
                        if await _concat_parts(part_files, seg_ts):
                            bg_tasks.append(asyncio.create_task(_remux_and_upload(seg_ts, seg_mp4, segment_num)))
                            uploaded_segments.append(segment_num)
                        else:
                            for pi, p in enumerate(part_files, 1):
                                if os.path.exists(p) and os.path.getsize(p) > 1024:
                                    p_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_fallback{pi:03d}.mp4")
                                    bg_tasks.append(asyncio.create_task(_remux_and_upload(p, p_mp4, pi)))
                            uploaded_segments.append(segment_num)
                        part_files = []
                    await asyncio.sleep(3)
                    part_num += 1
                    current_part = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_p{part_num:03d}.ts")
                    part_files.append(current_part)
                    process, used_proxy = await _start_recording(current_part, proxy_list)
                    if process is None:
                        logger.info(f"[LIVE:{task_id}] Cannot restart, stream ended")
                        break
                    # Wait up to 15s to see if it produces data or exits
                    for _ in range(5):
                        await asyncio.sleep(3)
                        if process.returncode is not None:
                            break
                        check_size = os.path.getsize(current_part) if os.path.exists(current_part) else 0
                        if check_size > 0:
                            break
                    if process.returncode is not None:
                        check_size = os.path.getsize(current_part) if os.path.exists(current_part) else 0
                        if check_size == 0:
                            logger.info(f"[LIVE:{task_id}] Restart produced no data, stream truly ended")
                            part_files = [p for p in part_files if p != current_part]
                            try: os.remove(current_part)
                            except: pass
                            break
                    # Stream still live — continue accumulating
                    logger.info(f"[LIVE:{task_id}] Stream still live, continuing (part {part_num})")
                    total_mb = _get_total_parts_size(part_files) / (1024*1024)
                    await live_status(f"\U0001f534 Recording: {channel_name} ({total_mb:.0f}MB)")
                    poll_count = 0
                    continue

            # Check cancel
            if task_id in cancelled_tasks:
                logger.info(f"[LIVE:{task_id}] Cancel signal")
                await _kill_process(process, task_id)
                # Don't discard here — let from-start also see the signal
                await update_status_msg("❌ Live recording cancelled.", force=True)
                return

            # Check stop & upload
            if task_id in stopped_tasks:
                logger.info(f"[LIVE:{task_id}] Stop & Upload signal")
                await _kill_process(process, task_id)
                # Don't discard here — let from-start also see the signal
                # Filter out empty/tiny parts before concat
                valid_parts = [p for p in part_files if os.path.exists(p) and os.path.getsize(p) > 1024]
                total_size = sum(os.path.getsize(p) for p in valid_parts)
                if total_size > 0 and valid_parts:
                    segment_num += 1
                    seg_ts = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.ts")
                    seg_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.mp4")
                    if await _concat_parts(valid_parts, seg_ts):
                        bg_tasks.append(asyncio.create_task(_remux_and_upload(seg_ts, seg_mp4, segment_num, is_final=True)))
                        uploaded_segments.append(segment_num)
                    elif len(valid_parts) == 1:
                        # Concat not needed for single file — just rename
                        os.rename(valid_parts[0], seg_ts)
                        bg_tasks.append(asyncio.create_task(_remux_and_upload(seg_ts, seg_mp4, segment_num, is_final=True)))
                        uploaded_segments.append(segment_num)
                    else:
                        # Concat failed — try uploading each part individually
                        logger.warning(f"[LIVE:{task_id}] Concat failed, uploading parts individually")
                        for pi, p in enumerate(valid_parts, 1):
                            p_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_part{pi:03d}.mp4")
                            bg_tasks.append(asyncio.create_task(_remux_and_upload(p, p_mp4, pi, is_final=(pi == len(valid_parts)))))
                        uploaded_segments.append(segment_num)
                # Clean up empty parts
                for p in part_files:
                    if p not in valid_parts and os.path.exists(p):
                        try: os.remove(p)
                        except: pass
                part_files = []
                break

            # Check from-start
            if task_id in fromstart_tasks and not fromstart_triggered:
                logger.info(f"[LIVE:{task_id}] 'From Start' triggered — killing streamlink, yt-dlp takes over")
                fromstart_triggered = True
                fromstart_tasks.discard(task_id)

                # 1. Kill streamlink
                await _kill_process(process, task_id)

                # 2. Upload whatever streamlink already recorded
                valid_parts = [p for p in part_files if os.path.exists(p) and os.path.getsize(p) > 1024]
                if valid_parts:
                    total_recorded = sum(os.path.getsize(p) for p in valid_parts)
                    logger.info(f"[LIVE:{task_id}] Uploading existing {total_recorded/(1024*1024):.1f}MB from streamlink before handoff")
                    segment_num += 1
                    seg_ts = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.ts")
                    seg_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.mp4")
                    if await _concat_parts(valid_parts, seg_ts):
                        bg_tasks.append(asyncio.create_task(_remux_and_upload(seg_ts, seg_mp4, segment_num)))
                        uploaded_segments.append(segment_num)
                    else:
                        for pi, p in enumerate(valid_parts, 1):
                            if os.path.exists(p) and os.path.getsize(p) > 1024:
                                p_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_fallback{pi:03d}.mp4")
                                bg_tasks.append(asyncio.create_task(_remux_and_upload(p, p_mp4, pi)))
                        uploaded_segments.append(segment_num)
                part_files = []

                # 3. Start from-start download (takes over entirely)
                await live_status(f"\U0001f534 {channel_name}\n⏪ From Start: yt-dlp taking over...")
                fromstart_task = asyncio.create_task(_download_from_start())
                bg_tasks.append(fromstart_task)
                break

            # Check total accumulated size
            total_size = _get_total_parts_size(part_files)

            if total_size >= SEGMENT_SIZE_BYTES:
                logger.info(f"[LIVE:{task_id}] Size limit {total_size/(1024*1024):.1f}MB — segmenting")

                # 1. Start new part BEFORE killing old one (zero gap)
                part_num += 1
                next_part = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_p{part_num:03d}.ts")
                new_process, new_proxy = await _start_recording(next_part, proxy_list)

                # 2. Kill old process
                await _kill_process(process, task_id)

                # 3. Concat current parts into segment, send to background upload
                segment_num += 1
                seg_ts = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.ts")
                seg_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.mp4")
                if await _concat_parts(part_files, seg_ts):
                    bg_tasks.append(asyncio.create_task(_remux_and_upload(seg_ts, seg_mp4, segment_num)))
                    uploaded_segments.append(segment_num)
                else:
                    for pi, p in enumerate(part_files, 1):
                        if os.path.exists(p) and os.path.getsize(p) > 1024:
                            p_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_fallback{pi:03d}.mp4")
                            bg_tasks.append(asyncio.create_task(_remux_and_upload(p, p_mp4, pi)))
                    uploaded_segments.append(segment_num)

                # 4. Reset parts for next segment
                part_files = [next_part]
                current_part = next_part

                if new_process is None:
                    logger.error(f"[LIVE:{task_id}] Failed to start next segment, ending")
                    break
                process = new_process
                used_proxy = new_proxy
                await live_status(f"\U0001f534 Recording: {channel_name} (Part {segment_num + 1})")
                poll_count = 0
                continue

            poll_count += 1
            if poll_count % 10 == 0:
                total_mb = total_size / (1024*1024)
                logger.info(f"[LIVE:{task_id}] Recording total={total_mb:.1f}MB parts={len(part_files)} polls={poll_count}")
                await live_status(f"\U0001f534 Recording: {channel_name} ({total_mb:.0f}MB)")

            await asyncio.sleep(3)

        # Upload remaining parts if any
        if part_files:
            valid_parts = [p for p in part_files if os.path.exists(p) and os.path.getsize(p) > 1024]
            total_size = sum(os.path.getsize(p) for p in valid_parts) if valid_parts else 0
            if total_size > 0 and valid_parts:
                segment_num += 1
                seg_ts = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.ts")
                seg_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_{segment_num:03d}.mp4")
                if await _concat_parts(valid_parts, seg_ts):
                    bg_tasks.append(asyncio.create_task(_remux_and_upload(seg_ts, seg_mp4, segment_num, is_final=True)))
                    uploaded_segments.append(segment_num)
                elif len(valid_parts) == 1:
                    os.rename(valid_parts[0], seg_ts)
                    bg_tasks.append(asyncio.create_task(_remux_and_upload(seg_ts, seg_mp4, segment_num, is_final=True)))
                    uploaded_segments.append(segment_num)
                else:
                    logger.warning(f"[LIVE:{task_id}] Final concat failed, uploading parts individually")
                    for pi, p in enumerate(valid_parts, 1):
                        p_mp4 = os.path.join(DOWNLOAD_DIR, f"live_{task_id}_part{pi:03d}.mp4")
                        bg_tasks.append(asyncio.create_task(_remux_and_upload(p, p_mp4, pi, is_final=(pi == len(valid_parts)))))
                    uploaded_segments.append(segment_num)
            # Clean up empty parts
            for p in part_files:
                if p not in valid_parts and os.path.exists(p):
                    try: os.remove(p)
                    except: pass

        # Wait for all background tasks
        if bg_tasks:
            logger.info(f"[LIVE:{task_id}] Waiting for {len(bg_tasks)} background tasks...")
            await update_status_msg(f"⬆️ Uploading {len(bg_tasks)} segment(s)...", force=True)
            await asyncio.gather(*bg_tasks, return_exceptions=True)
            bg_tasks.clear()

        logger.info(f"[LIVE:{task_id}] COMPLETE. Segments: {len(uploaded_segments)}")
        if status_msg:
            try: await tg_retry(status_msg.delete)
            except Exception as e:
                logger.warning(f"[LIVE:{task_id}] Could not delete status msg: {e}")

    except Exception as e:
        logger.error(f"[LIVE:{task_id}] UNHANDLED EXCEPTION: {e}", exc_info=True)
        try:
            await update_status_msg(f"\U0001f525 Live recording error: {e}", force=True)
        except Exception as e2:
            logger.error(f"[LIVE:{task_id}] Could not send error msg: {e2}", exc_info=True)
    finally:
        # Wait for any pending background tasks before cleanup
        try:
            if bg_tasks:
                logger.info(f"[LIVE:{task_id}] Waiting for {len(bg_tasks)} bg tasks before cleanup...")
                await asyncio.gather(*bg_tasks, return_exceptions=True)
        except NameError:
            pass
        stopped_tasks.discard(task_id)
        cancelled_tasks.discard(task_id)
        _cleanup_live_files(task_id)
        _cleanup_partial_downloads()
        _free_memory()


def _cleanup_live_files(task_id):
    """Remove any leftover live recording segments and temp files.
    When called with main task_id, skips fromstart files (they manage their own cleanup)."""
    is_fromstart = 'fromstart' in task_id
    for f in glob.glob(os.path.join(DOWNLOAD_DIR, f"live_{task_id}*")):
        if not is_fromstart and 'fromstart' in f:
            continue
        try: os.remove(f)
        except: pass

async def process_playlist_queue(application, playlist_queue):
    """Process playlist download queue SEQUENTIALLY to save space."""
    logger.info("Playlist queue processor started.")
    
    while True:
        task = await playlist_queue.get()
        try:
            status_msg_passed = None
            if len(task) == 5:
                chat_id, url, message_id, max_height, status_msg_passed = task
            else:
                chat_id, url, message_id, max_height = task
            
            task_id = f"pl_{chat_id}_{int(time.time())}"
            status_msg = status_msg_passed
            
            audio_only = (max_height in (-1, -2))
            audio_format = 'mp3' if max_height == -2 else 'm4a'
            
            last_edit_time = 0
            async def update_status_msg(text, force=True, show_cancel=True, send_new=False):
                nonlocal status_msg, last_edit_time
                now = time.time()
                if not force and (now - last_edit_time < 20):
                    return
                try:
                    reply_markup = None
                    if show_cancel:
                        keyboard = [[InlineKeyboardButton("❌ Cancel Playlist", callback_data=f"cancel:{task_id}")]]
                        reply_markup = InlineKeyboardMarkup(keyboard)
                        
                    if send_new and status_msg:
                        try:
                            await tg_retry(status_msg.delete)
                        except Exception as e:
                            logger.warning(f"Failed to delete old status msg: {e}")
                        status_msg = None

                    if status_msg:
                        if status_msg.text != text:
                            await tg_retry(status_msg.edit_text, text, reply_markup=reply_markup)
                            last_edit_time = now
                    else:
                        status_msg = await tg_retry(application.bot.send_message, chat_id=chat_id, text=text, reply_to_message_id=message_id, reply_markup=reply_markup)
                        last_edit_time = now
                except Exception as e:
                    logger.warning(f"Failed to update status: {e}")

            await update_status_msg("📋 Getting playlist info...")
            loop = asyncio.get_running_loop()

            try:
                cancel_event = asyncio.Event()

                async def _watch_cancel():
                    while not cancel_event.is_set():
                        if task_id in cancelled_tasks:
                            cancel_event.set()
                            return
                        await asyncio.sleep(1)

                watcher = asyncio.create_task(_watch_cancel())
                info_future = loop.run_in_executor(None, lambda: get_playlist_info(url))

                done, _ = await asyncio.wait(
                    [info_future, watcher],
                    timeout=60,
                    return_when=asyncio.FIRST_COMPLETED,
                )

                cancel_event.set()
                watcher.cancel()

                if task_id in cancelled_tasks:
                    await update_status_msg("❌ Playlist cancelled.")
                    cancelled_tasks.discard(task_id)
                    continue

                if info_future in done:
                    info = info_future.result()
                else:
                    await update_status_msg("❌ Timeout getting playlist info.")
                    continue

                entries = info.get('entries', [])
                playlist_title = info.get('title', 'Playlist')
                total_videos = len(entries)
                
                if total_videos == 0:
                    await update_status_msg("❌ No videos found in playlist.")
                    continue
                
                mode_str = f"Audio {audio_format.upper()}" if audio_only else f"{max_height}p"
                await update_status_msg(f"📋 Playlist: {playlist_title}\n🎬 Found {total_videos} videos.\n🚀 Starting sequential process ({mode_str})...")
                
                for i, entry in enumerate(entries):
                    if task_id in cancelled_tasks:
                        await update_status_msg("❌ Playlist cancelled.")
                        cancelled_tasks.discard(task_id)
                        _cleanup_partial_downloads()
                        break

                    v_url = entry['url']
                    v_title = entry['title']
                    v_id = entry.get('id') or v_url.split('v=')[-1].split('&')[0] if 'v=' in v_url else ''

                    await update_status_msg(f"🔄 Processing {i+1}/{total_videos}: {v_title[:30]}...", send_new=True)

                    def progress_cb(d):
                        if task_id in cancelled_tasks: raise Exception("Cancelled")
                        if d['status'] == 'downloading':
                            p = d.get('_percent_str', '0%')
                            asyncio.run_coroutine_threadsafe(update_status_msg(f"📋 Playlist: {i+1}/{total_videos}\n⬇️ {mode_str}: {p}", force=False), loop)

                    try:
                        file_path, title, video_id, thumb_path = await loop.run_in_executor(
                            None,
                            lambda: download_content(v_url, progress_cb, audio_only=audio_only, audio_format=audio_format, max_height=max_height, task_id=task_id, cancelled_tasks=cancelled_tasks)
                        )
                        await handle_upload(application, chat_id, file_path, title, v_url, audio_only, update_status_func=update_status_msg, reply_to_message_id=message_id, thumb_path=thumb_path)
                    except Exception as e:
                        logger.error(f"Failed for video {i+1}: {e}")
                        await application.bot.send_message(chat_id=chat_id, text=f"⚠️ Skipped {v_title[:30]}: {e}")
                        if v_id:
                            for f in glob.glob(os.path.join(DOWNLOAD_DIR, f"*{v_id}*")):
                                try: os.remove(f)
                                except: pass
                        continue

                
                await update_status_msg(f"✨ Playlist complete! Finished {total_videos} videos.", send_new=True)

            except Exception as e:
                await update_status_msg(f"❌ Failed to get playlist info: {e}")

        except Exception as e:
            logger.error(f"Playlist error: {e}")
            await update_status_msg(f"🔥 Error: {e}")
        finally:
            _cleanup_partial_downloads()
            playlist_queue.task_done()
            _free_memory()