irontide-session 1.0.1

#![allow(
    clippy::cast_possible_truncation,
    clippy::cast_precision_loss,
    clippy::cast_possible_wrap,
    clippy::cast_sign_loss,
    clippy::unchecked_time_subtraction,
    reason = "M175: piece arithmetic bounded by num_pieces (u32); resume-data fields follow wire-format integer widths"
)]

//! `TorrentActor` state machine, stats, resume data, and file management.
//!
//! This module contains an `impl TorrentActor` block with methods for:
//! - State transitions (`transition_state`)
//! - Storage operations (`handle_move_storage`, `handle_rename_file`)
//! - File completion tracking (`check_file_completion`, `fire_file_completed_alerts`)
//! - Progress computation (`compute_progress`)
//! - Error handling (`handle_clear_error`)
//! - File and flag status (`build_file_status`, `build_flags`, `apply_set_flags`, `apply_unset_flags`)
//! - Statistics (`make_stats`)
//! - Pause/resume (`handle_pause`, `handle_resume`)
//! - Piece verification (`verify_existing_pieces`, `handle_force_recheck`)
//! - Resume data (`build_resume_data`)
//! - File priorities (`handle_set_file_priority`)
//! - Seed ratio checking (`check_seed_ratio`)

use std::sync::Arc;

use tracing::info;

use crate::alert::{AlertKind, post_alert};
use crate::disk::DiskJobFlags;
use crate::peer_state::PeerSource;
use crate::piece_reservation::{AtomicPieceStates, BlockMaps, PieceState, StealCandidates};
use crate::torrent::{HashResult, TorrentActor, now_unix, relocate_files};
use crate::types::{PeerCommand, TorrentState, TorrentStats};

use irontide_core::FilePriority;
use irontide_storage::TorrentStorage;

impl TorrentActor {
    /// Transition to a new state, firing a `StateChanged` alert if different.
    pub(crate) fn transition_state(&mut self, new_state: TorrentState) {
        let prev = self.state;
        if prev == new_state {
            return;
        }

        let now = std::time::Instant::now();

        // Accumulate durations for the state we're LEAVING
        if let Some(since) = self.state_duration_since {
            let elapsed = now.duration_since(since).as_secs() as i64;
            match prev {
                TorrentState::Seeding => {
                    self.seeding_duration += elapsed;
                    self.finished_duration += elapsed;
                }
                TorrentState::Complete => {
                    self.finished_duration += elapsed;
                }
                _ => {}
            }
        }

        // Handle active_duration on pause/queue transitions
        if new_state == TorrentState::Paused || new_state == TorrentState::Queued {
            // Entering paused/queued: accumulate active time and clear timer
            if let Some(since) = self.active_since {
                self.active_duration += now.duration_since(since).as_secs() as i64;
            }
            self.active_since = None;
        } else if prev == TorrentState::Paused || prev == TorrentState::Queued {
            // Leaving paused/queued: restart active timer
            self.active_since = Some(now);
        }

        // Track first completion
        if matches!(new_state, TorrentState::Complete | TorrentState::Seeding)
            && !matches!(prev, TorrentState::Complete | TorrentState::Seeding)
            && self.completed_time == 0
        {
            self.completed_time = now_unix();
        }

        // v0.187.3 / Bug 8a: on transition INTO Seeding, request an immediate
        // choker run so the first leechers see Unchoke within milliseconds
        // rather than after the 10s `unchoke_interval` tick. Without this,
        // a freshly-completed torrent with interested peers can sit choked
        // for nearly a full interval, presenting as "no uploads after
        // finishing" in the dogfood report.
        if new_state == TorrentState::Seeding && prev != TorrentState::Seeding {
            self.force_immediate_choker_tick = true;
        }

        // Update state duration tracking
        self.state_duration_since = Some(now);
        self.need_save_resume = true;

        self.state = new_state;
        post_alert(
            &self.alert_tx,
            &self.alert_mask,
            AlertKind::StateChanged {
                info_hash: self.info_hash,
                prev_state: prev,
                new_state,
            },
        );
    }

    /// Handle `MoveStorage`: relocate data files, re-register storage.
    pub(crate) async fn handle_move_storage(
        &mut self,
        new_path: std::path::PathBuf,
    ) -> crate::Result<()> {
        self.moving_storage = true;

        let Some(meta) = self.meta.as_ref() else {
            self.moving_storage = false;
            return Err(crate::Error::Config(
                "cannot move storage: metadata not available".into(),
            ));
        };

        let file_paths: Vec<std::path::PathBuf> = meta
            .info
            .files()
            .iter()
            .map(|f| f.path.iter().collect::<std::path::PathBuf>())
            .collect();
        let file_lengths: Vec<u64> = meta.info.files().iter().map(|f| f.length).collect();
        // files() already includes the torrent name as the first path component,
        // so src/dst base is just the download directory — no extra join with name.
        let src_base = self.config.download_dir.clone();
        let dst_base = new_path.clone();

        // Relocate files on a blocking thread to avoid starving the async runtime
        let src = src_base.clone();
        let dst = dst_base.clone();
        let paths = file_paths.clone();
        let result = tokio::task::spawn_blocking(move || relocate_files(&src, &dst, &paths))
            .await
            .map_err(|e| crate::Error::Io(std::io::Error::other(e)))
            .and_then(|r| r.map_err(crate::Error::Io));
        if let Err(e) = result {
            self.moving_storage = false;
            return Err(e);
        }

        // Unregister old storage
        self.disk_manager.unregister_torrent(self.info_hash).await;

        // Create new storage at destination
        let Some(lengths) = self.lengths.clone() else {
            self.moving_storage = false;
            return Err(crate::Error::Config("lengths not available".into()));
        };
        let prealloc_mode = self.config.preallocate_mode.unwrap_or_else(|| {
            irontide_storage::PreallocateMode::from(
                self.config.storage_mode == irontide_core::StorageMode::Full,
            )
        });
        let storage: Arc<dyn TorrentStorage> = match irontide_storage::FilesystemStorage::new(
            &new_path,
            file_paths,
            file_lengths,
            lengths,
            Some(&self.file_priorities),
            prealloc_mode,
            self.config.filesystem_direct_io,
        ) {
            Ok(s) => Arc::new(s),
            Err(e) => {
                self.moving_storage = false;
                return Err(e.into());
            }
        };

        // Re-register with disk manager
        self.disk = Some(
            self.disk_manager
                .register_torrent(self.info_hash, storage)
                .await,
        );

        // Update download dir
        self.config.download_dir.clone_from(&new_path);

        // Fire alert
        post_alert(
            &self.alert_tx,
            &self.alert_mask,
            AlertKind::StorageMoved {
                info_hash: self.info_hash,
                new_path,
            },
        );

        self.moving_storage = false;
        Ok(())
    }

    /// Handle `RenameFile`: rename a single file within the torrent on disk.
    pub(crate) async fn handle_rename_file(
        &mut self,
        file_index: usize,
        new_name: String,
    ) -> crate::Result<()> {
        let Some(meta) = self.meta.as_ref() else {
            return Err(crate::Error::Config(
                "cannot rename file: metadata not available".into(),
            ));
        };

        let files = meta.info.files();
        if file_index >= files.len() {
            return Err(crate::Error::Config(format!(
                "file index {file_index} out of range (torrent has {} files)",
                files.len()
            )));
        }

        // Compute the old relative path (files() includes torrent name as first component)
        let old_rel: std::path::PathBuf = files[file_index].path.iter().collect();
        let old_path = self.config.download_dir.join(&old_rel);

        // Build new relative path: same parent directory, new filename
        let new_rel = if let Some(parent) = old_rel.parent() {
            parent.join(&new_name)
        } else {
            std::path::PathBuf::from(&new_name)
        };
        let new_path = self.config.download_dir.join(&new_rel);

        // Perform the rename on a blocking thread
        let src = old_path.clone();
        let dst = new_path.clone();
        tokio::task::spawn_blocking(move || -> std::io::Result<()> {
            if let Some(parent) = dst.parent() {
                std::fs::create_dir_all(parent)?;
            }
            std::fs::rename(&src, &dst)
        })
        .await
        .map_err(|e| crate::Error::Io(std::io::Error::other(e)))
        .and_then(|r| r.map_err(crate::Error::Io))?;

        // Fire FileRenamed alert
        post_alert(
            &self.alert_tx,
            &self.alert_mask,
            AlertKind::FileRenamed {
                info_hash: self.info_hash,
                index: file_index,
                new_path: new_path.clone(),
            },
        );

        Ok(())
    }

    /// Check if the just-completed piece finishes any file, and fire `FileCompleted` alerts.
    ///
    /// M116: Uses pre-computed `cached_files` mapping instead of allocating
    /// `meta.info.files()` on every verified piece.
    pub(crate) fn check_file_completion(&self, piece_index: u32) {
        let Some(cached) = self.cached_files.as_ref() else {
            return;
        };
        let bitfield = match self.chunk_tracker.as_ref() {
            Some(ct) => ct.bitfield(),
            None => return,
        };

        for entry in &cached.entries {
            // Skip files that don't contain this piece
            if piece_index < entry.first_piece || piece_index > entry.last_piece {
                continue;
            }

            // Check if ALL pieces for this file are complete
            let all_complete = (entry.first_piece..=entry.last_piece).all(|p| bitfield.get(p));

            if all_complete {
                post_alert(
                    &self.alert_tx,
                    &self.alert_mask,
                    AlertKind::FileCompleted {
                        info_hash: self.info_hash,
                        file_index: entry.index,
                    },
                );
            }
        }
    }

    /// Compute download progress metrics.
    ///
    /// Returns `(total, total_done, total_wanted, total_wanted_done, progress, progress_ppm)`.
    pub(crate) fn compute_progress(&self) -> (u64, u64, u64, u64, f32, u32) {
        let Some(lengths) = &self.lengths else {
            return (0, 0, 0, 0, 0.0, 0);
        };

        let total = lengths.total_length();

        let bitfield = self
            .chunk_tracker
            .as_ref()
            .map(irontide_storage::ChunkTracker::bitfield);

        let mut total_done: u64 = 0;
        let mut total_wanted: u64 = 0;
        let mut total_wanted_done: u64 = 0;

        for idx in 0..self.num_pieces {
            let piece_bytes = u64::from(lengths.piece_size(idx));
            let have = bitfield.as_ref().is_some_and(|bf| bf.get(idx));

            if have {
                total_done += piece_bytes;
            }

            if self.wanted_pieces.get(idx) {
                total_wanted += piece_bytes;
                if have {
                    total_wanted_done += piece_bytes;
                }
            }
        }

        let progress = if total_wanted == 0 {
            1.0
        } else {
            total_wanted_done as f32 / total_wanted as f32
        };
        let progress_ppm = (progress * 1_000_000.0) as u32;

        (
            total,
            total_done,
            total_wanted,
            total_wanted_done,
            progress,
            progress_ppm,
        )
    }

    /// Clear the error state. If the torrent was paused and had an error, resume it.
    pub(crate) async fn handle_clear_error(&mut self) {
        let had_error = !self.error.is_empty();
        self.error = String::new();
        self.error_file = -1;
        // If we were paused/queued and had an error, resume
        if had_error && matches!(self.state, TorrentState::Paused | TorrentState::Queued) {
            self.handle_resume().await;
        }
    }

    /// Build per-file status based on the current torrent state.
    pub(crate) fn build_file_status(&self) -> Vec<crate::types::FileStatus> {
        let num_files = self.file_priorities.len();
        let (open, mode) = match self.state {
            TorrentState::Seeding => (true, crate::types::FileMode::ReadOnly),
            TorrentState::Downloading
            | TorrentState::Checking
            | TorrentState::FetchingMetadata
            | TorrentState::Complete
            | TorrentState::Sharing => (true, crate::types::FileMode::ReadWrite),
            TorrentState::Paused | TorrentState::Queued | TorrentState::Stopped => {
                (false, crate::types::FileMode::Closed)
            }
        };
        vec![crate::types::FileStatus { open, mode }; num_files]
    }

    /// Build the current `TorrentFlags` from actor state.
    pub(crate) fn build_flags(&self) -> crate::types::TorrentFlags {
        let mut flags = crate::types::TorrentFlags::empty();
        if self.state == TorrentState::Paused {
            flags |= crate::types::TorrentFlags::PAUSED;
        }
        // auto_managed is session-level; torrent actor doesn't track it.
        // We leave AUTO_MANAGED unset at the torrent level.
        if self.config.sequential_download {
            flags |= crate::types::TorrentFlags::SEQUENTIAL_DOWNLOAD;
        }
        if self.config.super_seeding {
            flags |= crate::types::TorrentFlags::SUPER_SEEDING;
        }
        if self.state == TorrentState::Seeding || matches!(self.state, TorrentState::Complete) {
            flags |= crate::types::TorrentFlags::UPLOAD_ONLY;
        }
        flags
    }

    /// Apply `set_flags`: enable the specified flags.
    pub(crate) async fn apply_set_flags(&mut self, flags: crate::types::TorrentFlags) {
        if flags.contains(crate::types::TorrentFlags::PAUSED) && self.state != TorrentState::Paused
        {
            self.handle_pause().await;
        }
        if flags.contains(crate::types::TorrentFlags::SEQUENTIAL_DOWNLOAD) {
            self.config.sequential_download = true;
        }
        if flags.contains(crate::types::TorrentFlags::SUPER_SEEDING) {
            self.config.super_seeding = true;
            if self.super_seed.is_none() {
                self.super_seed = Some(crate::super_seed::SuperSeedState::new());
            }
        }
        // AUTO_MANAGED and UPLOAD_ONLY are session-level; no-op at torrent level.
    }

    /// Apply `unset_flags`: disable the specified flags.
    pub(crate) async fn apply_unset_flags(&mut self, flags: crate::types::TorrentFlags) {
        if flags.contains(crate::types::TorrentFlags::PAUSED) && self.state == TorrentState::Paused
        {
            self.handle_resume().await;
        }
        if flags.contains(crate::types::TorrentFlags::SEQUENTIAL_DOWNLOAD) {
            self.config.sequential_download = false;
        }
        if flags.contains(crate::types::TorrentFlags::SUPER_SEEDING) {
            self.config.super_seeding = false;
            self.super_seed = None;
        }
        // AUTO_MANAGED and UPLOAD_ONLY are session-level; no-op at torrent level.
    }

    pub(crate) fn make_stats(&self) -> TorrentStats {
        // ── Single pass over peers ──
        let mut num_seeds = 0usize;
        let mut num_uploads = 0usize;
        let mut download_rate_sum: u64 = 0;
        let mut upload_rate_sum: u64 = 0;
        let mut peers_by_source = std::collections::HashMap::new();

        for peer in self.peers.values() {
            *peers_by_source.entry(peer.source).or_insert(0) += 1;
            download_rate_sum += peer.download_rate;
            upload_rate_sum += peer.upload_rate;
            if self.num_pieces > 0 && peer.bitfield.count_ones() == self.num_pieces {
                num_seeds += 1;
            }
            if !peer.am_choking {
                num_uploads += 1;
            }
        }

        // ── Tracker info (scrape data + current tracker) ──
        let tracker_list = self.tracker_manager.tracker_list();
        let mut num_complete: i32 = -1;
        let mut num_incomplete: i32 = -1;
        let mut current_tracker = String::new();

        for ti in &tracker_list {
            if num_complete == -1
                && let Some(s) = ti.seeders
            {
                num_complete = s as i32;
            }
            if num_incomplete == -1
                && let Some(l) = ti.leechers
            {
                num_incomplete = l as i32;
            }
            if current_tracker.is_empty()
                && matches!(ti.status, crate::tracker_manager::TrackerStatus::Working)
            {
                current_tracker.clone_from(&ti.url);
            }
        }

        // ── Progress ──
        let pieces_have = self
            .chunk_tracker
            .as_ref()
            .map_or(0, |ct| ct.bitfield().count_ones());
        let (total, total_done, total_wanted, total_wanted_done, progress, progress_ppm) =
            self.compute_progress();

        // ── Distributed copies ──
        let (distributed_full_copies, distributed_fraction, distributed_copies) =
            self.distributed_copies();

        // ── Active duration (include current active stint) ──
        let active_duration = self.active_duration
            + self
                .active_since
                .map_or(0, |since| since.elapsed().as_secs() as i64);

        // ── Finished duration (include current stint if Complete or Seeding) ──
        let finished_duration = self.finished_duration
            + self
                .state_duration_since
                .filter(|_| matches!(self.state, TorrentState::Complete | TorrentState::Seeding))
                .map_or(0, |since| since.elapsed().as_secs() as i64);

        // ── Seeding duration (include current stint if Seeding) ──
        let seeding_duration = self.seeding_duration
            + self
                .state_duration_since
                .filter(|_| self.state == TorrentState::Seeding)
                .map_or(0, |since| since.elapsed().as_secs() as i64);

        // ── Name ──
        let name = self
            .meta
            .as_ref()
            .map(|m| m.info.name.clone())
            .unwrap_or_default();

        // ── Block size ──
        let block_size = self
            .lengths
            .as_ref()
            .map_or(16384, irontide_core::Lengths::chunk_size);

        // ── M170: category / created_by / creation_date / piece_size ──
        //
        // `created_by` and `creation_date` are sourced directly from the
        // parsed metadata on every stats call. For magnet-added torrents
        // they appear automatically once metadata resolves and
        // `self.meta` gets populated (no extra dirty-bit plumbing needed).
        // `piece_size` comes from `Lengths::piece_length()` once lengths
        // are known; 0 until metadata resolves. The category label lives
        // on `self.config.category` so that the torrent actor owns its
        // current assignment (updates flow via `SetCategory` in a future
        // milestone).
        let created_by = self.meta.as_ref().and_then(|m| m.created_by.clone());
        let creation_date = self.meta.as_ref().and_then(|m| m.creation_date);
        let piece_size = self
            .lengths
            .as_ref()
            .map_or(0, irontide_core::Lengths::piece_length);

        TorrentStats {
            // ── Original 9 fields (unchanged) ──
            state: self.state,
            downloaded: self.downloaded,
            uploaded: self.uploaded,
            pieces_have,
            pieces_total: self.num_pieces,
            peers_connected: self.peers.len(),
            peers_available: 0, // M107: discovery pool is in adder task channel
            checking_progress: self.checking_progress,
            peers_by_source,

            // ── Identity ──
            info_hashes: self.info_hashes.clone(),
            name,

            // ── State flags ──
            has_metadata: self.meta.is_some(),
            is_seeding: self.state == TorrentState::Seeding,
            is_finished: matches!(self.state, TorrentState::Complete | TorrentState::Seeding),
            is_paused: self.state == TorrentState::Paused,
            is_queued: self.state == TorrentState::Queued,
            auto_managed: false, // session fills this
            sequential_download: self.config.sequential_download,
            super_seeding: self.config.super_seeding,
            user_seed_mode: self.user_seed_mode,
            user_forced: self.user_forced,
            seed_ratio_override: self.config.seed_ratio_limit,
            has_incoming: self.has_incoming,
            need_save_resume: self.need_save_resume,
            moving_storage: self.moving_storage,

            // ── Progress ──
            progress,
            progress_ppm,
            total_done,
            total,
            total_wanted_done,
            total_wanted,
            block_size,

            // ── Transfer (session counters) ──
            total_download: self.total_download,
            total_upload: self.total_upload,
            total_payload_download: self.downloaded,
            total_payload_upload: self.uploaded,
            total_failed_bytes: self.total_failed_bytes,
            total_redundant_bytes: self.total_redundant_bytes,

            // ── Transfer (all-time = session for now, no persistence yet) ──
            all_time_download: self.total_download,
            all_time_upload: self.total_upload,

            // ── Rates ──
            download_rate: download_rate_sum,
            upload_rate: upload_rate_sum,
            download_payload_rate: download_rate_sum,
            upload_payload_rate: upload_rate_sum,

            // ── Connection details ──
            num_peers: self.peers.len(),
            num_seeds,
            num_complete,
            num_incomplete,
            list_seeds: num_seeds,
            list_peers: self.peers.len(), // M107: discovery pool is in adder task
            connect_candidates: 0,        // M107: discovery pool is in adder task
            num_connections: self.peers.len(),
            num_uploads,
            unique_peers_attempted: self
                .peer_states
                .as_ref()
                .map_or(0, |ps| u64::from(ps.stats.snapshot().known)),
            pipeline: self.peer_states.as_ref().map(|ps| ps.stats.snapshot()),
            choke_rotations: self.choke_rotations,
            piece_steals: self.piece_steals,
            holepunch_relayed: self.holepunch_relayed,
            dispatch_pieces_queued: self
                .piece_tracker
                .as_ref()
                .map_or(0, crate::piece_reservation::PieceTracker::queue_count),
            dispatch_pieces_inflight: self
                .piece_tracker
                .as_ref()
                .map_or(0, |pt| pt.inflight_count() as u32),

            // ── Limits ──
            connections_limit: self.effective_max_connections(),
            uploads_limit: self.choker.unchoke_slots(),

            // ── Distributed copies ──
            distributed_full_copies,
            distributed_fraction,
            distributed_copies,

            // ── Tracker ──
            current_tracker,
            announcing_to_trackers: !tracker_list.is_empty(),
            announcing_to_lsd: false, // LSD not yet implemented
            announcing_to_dht: self.dht_peers_rx.is_some(),

            // ── Timestamps ──
            added_time: self.added_time,
            completed_time: self.completed_time,
            last_seen_complete: self.last_seen_complete,
            last_upload: self.last_upload,
            last_download: self.last_download,

            // ── Durations ──
            active_duration,
            finished_duration,
            seeding_duration,

            // ── Storage ──
            save_path: self.config.download_dir.to_string_lossy().into_owned(),

            // ── Queue (session fills this) ──
            queue_position: -1,

            // ── Error ──
            error: self.error.clone(),
            error_file: self.error_file,

            // ── M170 ──
            category: self.config.category.clone(),
            created_by,
            creation_date,
            piece_size,

            // ── M171 ──
            tags: self.config.tags.clone(),
        }
    }

    pub(crate) async fn handle_pause(&mut self) {
        if self.state == TorrentState::Paused || self.state == TorrentState::Stopped {
            return;
        }
        let prev_state = self.state;
        self.transition_state(TorrentState::Paused);
        post_alert(
            &self.alert_tx,
            &self.alert_mask,
            AlertKind::TorrentPaused {
                info_hash: self.info_hash,
            },
        );
        // Disconnect all peers (non-blocking — peer may already be dead)
        for peer in self.peers.values() {
            let _ = peer.cmd_tx.try_send(PeerCommand::Shutdown);
        }
        self.peers.clear();
        // Announce Stopped to trackers (with timeout to prevent hang)
        if prev_state == TorrentState::Downloading
            || prev_state == TorrentState::Seeding
            || prev_state == TorrentState::Complete
        {
            let left = self.calculate_left();
            let _ = tokio::time::timeout(
                std::time::Duration::from_secs(3),
                self.tracker_manager
                    .announce_stopped(self.uploaded, self.downloaded, left),
            )
            .await;
        }
    }

    /// Queue a torrent via auto-manage. Unlike `handle_pause`, this does NOT
    /// announce Stopped to trackers (queued torrents auto-resume soon) and does
    /// NOT set the PAUSED flag (so F4 guards don't trap it).
    pub(crate) fn handle_queue(&mut self) {
        if matches!(
            self.state,
            TorrentState::Paused | TorrentState::Queued | TorrentState::Stopped
        ) {
            return;
        }
        self.transition_state(TorrentState::Queued);
        post_alert(
            &self.alert_tx,
            &self.alert_mask,
            AlertKind::TorrentPaused {
                info_hash: self.info_hash,
            },
        );
        for peer in self.peers.values() {
            let _ = peer.cmd_tx.try_send(PeerCommand::Shutdown);
        }
        self.peers.clear();
    }

    /// M159: Toggle user-requested seed-only mode.
    ///
    /// When `enabled` is `true`, stops scheduling new block requests and
    /// cancels every in-flight block request, but keeps peers connected and
    /// continues serving uploads. Reverts to normal piece scheduling on
    /// `false`. Idempotent.
    pub(crate) fn handle_set_seed_mode(&mut self, enabled: bool) {
        if self.user_seed_mode == enabled {
            // Idempotent — even if natural state changed, nothing new to do.
            return;
        }

        self.user_seed_mode = enabled;

        if enabled {
            // Flip the choker into seed mode (affects choke decisions only).
            self.choker.set_seed_mode(true);

            // 1. Cancel every in-flight request on every peer. Mirror the
            //    per-block Cancel pattern used by the M149 steal path
            //    (`torrent_peers::run_scored_turnover`): send one
            //    `PeerCommand::Cancel` per outstanding block, clear the
            //    peer's local `pending_requests` tracker, then send
            //    `StopRequesting` so the peer's requester loop drops its
            //    dispatch state and awaits a fresh `StartRequesting`.
            for peer in self.peers.values_mut() {
                let cancels: Vec<(u32, u32, u32)> = peer.pending_requests.iter().collect();
                for (index, begin, length) in cancels {
                    let _ = peer.cmd_tx.try_send(PeerCommand::Cancel {
                        index,
                        begin,
                        length,
                    });
                }
                peer.pending_requests.clear();
                let _ = peer.cmd_tx.try_send(PeerCommand::StopRequesting);
            }

            // 2. Release every Reserved/Endgame piece so the reservation
            //    state isn't poisoned when seed mode is turned off again.
            //    Also clear BlockMaps for those pieces so block-level
            //    stealing (M103) starts from a clean slate on resume.
            if let Some(ref atomic_states) = self.atomic_states {
                let num_pieces = self.num_pieces;
                let lengths = self.lengths.clone();
                let block_maps = self.block_maps.clone();
                let steal_candidates = self.steal_candidates.clone();
                for piece in 0..num_pieces {
                    let st = atomic_states.get(piece);
                    if matches!(st, PieceState::Reserved | PieceState::Endgame) {
                        atomic_states.release(piece);
                        if let (Some(bm), Some(l)) = (&block_maps, &lengths) {
                            bm.clear(piece, l.chunks_in_piece(piece));
                        }
                        if let Some(ref sc) = steal_candidates {
                            sc.remove(piece);
                        }
                        if let Some(slot) = self.piece_owner.get_mut(piece as usize) {
                            *slot = None;
                        }
                        if let Some(slot) = self.inflight_started.get_mut(piece as usize) {
                            *slot = None;
                        }
                        // M187: Return piece to queue.
                        if let Some(ref mut pt) = self.piece_tracker {
                            pt.mark_piece_hash_failed(piece);
                        }
                    }
                }
            }

            // 3. End-game (if active) is no longer meaningful until we
            //    resume downloading — drain its pending block set.
            self.end_game.deactivate();

            // 4. Rebuild availability snapshot next tick so peers that
            //    later resume dispatch see the cleaned-up Available state.
        } else {
            // Disabling: restore the choker's seed flag to reflect the
            // *natural* completion state. `set_seed_mode(true)` is kept
            // only if the torrent is actually fully downloaded.
            let naturally_seeding =
                matches!(self.state, TorrentState::Seeding | TorrentState::Complete)
                    || self
                        .chunk_tracker
                        .as_ref()
                        .is_some_and(|ct| ct.bitfield().count_ones() == self.num_pieces);
            self.choker.set_seed_mode(naturally_seeding);

            // Re-issue StartRequesting to every connected peer so their
            // requester loops exit the idle phase and begin dispatching
            // again. Uses the same plumbing as the post-metadata fan-out.
            if let Some(notify) = &self.reservation_notify
                && let Some(ref lengths) = self.lengths
            {
                for peer in self.peers.values() {
                    let _ = peer.cmd_tx.try_send(PeerCommand::StartRequesting {
                        piece_notify: Arc::clone(notify),
                        disk_handle: self.disk.clone(),
                        write_error_tx: self.write_error_tx.clone(),
                        lengths: lengths.clone(),
                    });
                }
            }

            // Force a fresh snapshot rebuild on the next tick so the
            // resumed peers pick up the (now-available) pieces.
        }
    }

    pub(crate) async fn handle_resume(&mut self) {
        if !matches!(self.state, TorrentState::Paused | TorrentState::Queued) {
            return;
        }
        // Determine appropriate state
        if self.config.share_mode {
            self.transition_state(TorrentState::Sharing);
        } else if self.num_pieces == 0 && self.chunk_tracker.is_none() {
            self.transition_state(TorrentState::FetchingMetadata);
        } else if let Some(ref ct) = self.chunk_tracker
            && ct.bitfield().count_ones() == self.num_pieces
        {
            self.transition_state(TorrentState::Seeding);
            self.choker.set_seed_mode(true);
        } else {
            self.transition_state(TorrentState::Downloading);
            self.choker.set_seed_mode(false);
        }
        post_alert(
            &self.alert_tx,
            &self.alert_mask,
            AlertKind::TorrentResumed {
                info_hash: self.info_hash,
            },
        );
        // Re-announce Started
        let left = self.calculate_left();
        let result = self
            .tracker_manager
            .announce(
                irontide_tracker::AnnounceEvent::Started,
                self.uploaded,
                self.downloaded,
                left,
            )
            .await;
        self.fire_tracker_alerts(&result.outcomes);
        if !result.peers.is_empty() {
            self.handle_add_peers(result.peers, PeerSource::Tracker);
        }
    }

    pub(crate) async fn verify_existing_pieces(&mut self) {
        let disk = match &self.disk {
            Some(d) => d.clone(),
            None => return,
        };
        let Some(meta) = self.meta.clone() else {
            return;
        };

        self.transition_state(TorrentState::Checking);
        self.checking_progress = 0.0;

        let mut verified_count = 0u32;
        let total = self.num_pieces;

        // M205: fast file-size pre-scan to skip pieces with missing/short files.
        // When no files are found at all, fall through to full verification
        // (data may live in non-filesystem storage).
        let file_infos = meta.info.files();
        let scan = crate::verify_before_download::quick_file_scan(
            &file_infos,
            meta.info.piece_length,
            total,
            &self.config.download_dir,
        );
        let scan = if scan.files_found == 0 {
            crate::verify_before_download::ScanResult {
                candidates: vec![true; total as usize],
                candidate_count: total,
                total_pieces: total,
                files_found: 0,
                files_total: scan.files_total,
            }
        } else {
            if scan.candidate_count < total {
                info!(
                    candidates = scan.candidate_count,
                    total,
                    files_found = scan.files_found,
                    files_total = scan.files_total,
                    "pre-scan: verifying candidate pieces only"
                );
            }
            scan
        };

        if self.version == irontide_core::TorrentVersion::V2Only {
            // V2Only: use SHA-256 Merkle block verification (sequential, needs &mut self)
            for piece in 0..total {
                if !scan.candidates[piece as usize] {
                    self.checking_progress = (piece + 1) as f32 / total as f32;
                    post_alert(
                        &self.alert_tx,
                        &self.alert_mask,
                        AlertKind::CheckingProgress {
                            info_hash: self.info_hash,
                            progress: self.checking_progress,
                        },
                    );
                    continue;
                }

                let result = self.run_v2_block_verification(piece).await;
                if matches!(result, HashResult::Passed) {
                    if let Some(ref mut ct) = self.chunk_tracker {
                        ct.mark_verified(piece);
                    }
                    verified_count += 1;
                }

                self.checking_progress = (piece + 1) as f32 / total as f32;
                post_alert(
                    &self.alert_tx,
                    &self.alert_mask,
                    AlertKind::CheckingProgress {
                        info_hash: self.info_hash,
                        progress: self.checking_progress,
                    },
                );
            }
        } else {
            // V1Only / Hybrid: use concurrent SHA-1 piece verification
            let max_concurrent = self.config.hashing_threads.max(1);
            let mut checked_count = 0u32;
            let pieces_to_check = scan.candidate_count;
            let mut in_flight = tokio::task::JoinSet::new();
            let mut next_piece = 0u32;

            // Seed the pipeline (skip non-candidate pieces)
            while next_piece < total && in_flight.len() < max_concurrent {
                if scan.candidates[next_piece as usize]
                    && let Some(expected) = meta.info.piece_hash(next_piece as usize)
                {
                    let d = disk.clone();
                    let piece = next_piece;
                    in_flight.spawn(async move {
                        let valid = d
                            .verify_piece(piece, expected, DiskJobFlags::empty())
                            .await
                            .unwrap_or(false);
                        (piece, valid)
                    });
                }
                next_piece += 1;
            }

            // Process completions, refill pipeline
            while let Some(result) = in_flight.join_next().await {
                if let Ok((piece, valid)) = result {
                    checked_count += 1;
                    if valid {
                        if let Some(ref mut ct) = self.chunk_tracker {
                            ct.mark_verified(piece);
                        }
                        verified_count += 1;
                    }

                    // Update progress based on candidate count
                    self.checking_progress = if pieces_to_check > 0 {
                        checked_count as f32 / pieces_to_check as f32
                    } else {
                        1.0
                    };
                    post_alert(
                        &self.alert_tx,
                        &self.alert_mask,
                        AlertKind::CheckingProgress {
                            info_hash: self.info_hash,
                            progress: self.checking_progress,
                        },
                    );
                }

                // Refill pipeline (skip non-candidate pieces)
                while next_piece < total && in_flight.len() < max_concurrent {
                    if scan.candidates[next_piece as usize]
                        && let Some(expected) = meta.info.piece_hash(next_piece as usize)
                    {
                        let d = disk.clone();
                        let piece = next_piece;
                        in_flight.spawn(async move {
                            let valid = d
                                .verify_piece(piece, expected, DiskJobFlags::empty())
                                .await
                                .unwrap_or(false);
                            (piece, valid)
                        });
                    }
                    next_piece += 1;
                }
            }
        }

        // Fire TorrentChecked alert
        self.checking_progress = 0.0;
        post_alert(
            &self.alert_tx,
            &self.alert_mask,
            AlertKind::TorrentChecked {
                info_hash: self.info_hash,
                pieces_have: verified_count,
                pieces_total: total,
            },
        );

        if verified_count > 0 {
            info!(verified_count, total, "resumed with existing pieces");
        }

        if self.config.share_mode {
            self.transition_state(TorrentState::Sharing);
        } else if verified_count == self.num_pieces {
            self.transition_state(TorrentState::Seeding);
            self.choker.set_seed_mode(true);
            info!("all pieces verified, starting as seeder");
        } else {
            self.transition_state(TorrentState::Downloading);
            self.choker.set_seed_mode(false);
        }

        // Fire FileCompleted alerts for any files that are fully verified
        self.fire_file_completed_alerts();
    }

    /// Verify existing pieces interactively, draining read-only commands
    /// (Stats, Info, etc.) from the command channel between verification
    /// batches so the GUI poll loop can observe Checking state and progress.
    ///
    /// This is the force-recheck counterpart to [`verify_existing_pieces`],
    /// which blocks the actor for the entire duration. Here we process pending
    /// commands after each `JoinSet` completion so the actor stays responsive.
    pub(crate) async fn verify_existing_pieces_interactive(&mut self) {
        let disk = match &self.disk {
            Some(d) => d.clone(),
            None => return,
        };
        let Some(meta) = self.meta.clone() else {
            return;
        };

        self.transition_state(TorrentState::Checking);
        self.checking_progress = 0.0;

        let mut verified_count = 0u32;
        let total = self.num_pieces;

        if self.version == irontide_core::TorrentVersion::V2Only {
            for piece in 0..total {
                let result = self.run_v2_block_verification(piece).await;
                if matches!(result, HashResult::Passed) {
                    if let Some(ref mut ct) = self.chunk_tracker {
                        ct.mark_verified(piece);
                    }
                    verified_count += 1;
                }

                self.checking_progress = (piece + 1) as f32 / total as f32;
                post_alert(
                    &self.alert_tx,
                    &self.alert_mask,
                    AlertKind::CheckingProgress {
                        info_hash: self.info_hash,
                        progress: self.checking_progress,
                    },
                );

                // Drain read-only commands so the poll loop stays responsive.
                self.drain_read_commands_during_check();
            }
        } else {
            let max_concurrent = self.config.hashing_threads.max(1);
            let mut checked_count = 0u32;
            let mut in_flight = tokio::task::JoinSet::new();
            let mut next_piece = 0u32;

            while next_piece < total && in_flight.len() < max_concurrent {
                if let Some(expected) = meta.info.piece_hash(next_piece as usize) {
                    let d = disk.clone();
                    let piece = next_piece;
                    in_flight.spawn(async move {
                        let valid = d
                            .verify_piece(piece, expected, DiskJobFlags::empty())
                            .await
                            .unwrap_or(false);
                        (piece, valid)
                    });
                }
                next_piece += 1;
            }

            while let Some(result) = in_flight.join_next().await {
                if let Ok((piece, valid)) = result {
                    checked_count += 1;
                    if valid {
                        if let Some(ref mut ct) = self.chunk_tracker {
                            ct.mark_verified(piece);
                        }
                        verified_count += 1;
                    }

                    self.checking_progress = checked_count as f32 / total as f32;
                    post_alert(
                        &self.alert_tx,
                        &self.alert_mask,
                        AlertKind::CheckingProgress {
                            info_hash: self.info_hash,
                            progress: self.checking_progress,
                        },
                    );
                }

                while next_piece < total && in_flight.len() < max_concurrent {
                    if let Some(expected) = meta.info.piece_hash(next_piece as usize) {
                        let d = disk.clone();
                        let piece = next_piece;
                        in_flight.spawn(async move {
                            let valid = d
                                .verify_piece(piece, expected, DiskJobFlags::empty())
                                .await
                                .unwrap_or(false);
                            (piece, valid)
                        });
                    }
                    next_piece += 1;
                }

                // Drain read-only commands so the poll loop stays responsive.
                self.drain_read_commands_during_check();
            }
        }

        self.checking_progress = 0.0;
        post_alert(
            &self.alert_tx,
            &self.alert_mask,
            AlertKind::TorrentChecked {
                info_hash: self.info_hash,
                pieces_have: verified_count,
                pieces_total: total,
            },
        );

        if verified_count > 0 {
            info!(
                verified_count,
                total, "recheck: resumed with existing pieces"
            );
        }

        if self.config.share_mode {
            self.transition_state(TorrentState::Sharing);
        } else if verified_count == self.num_pieces {
            self.transition_state(TorrentState::Seeding);
            self.choker.set_seed_mode(true);
            info!("recheck: all pieces verified, starting as seeder");
        } else {
            self.transition_state(TorrentState::Downloading);
            self.choker.set_seed_mode(false);
        }

        self.fire_file_completed_alerts();
    }

    /// Drain pending read-only commands from the command channel during
    /// a long-running operation (e.g. recheck). Answers Stats and Info
    /// queries immediately so the GUI poll loop isn't blocked. Other
    /// commands are re-sent back into the channel for later processing.
    fn drain_read_commands_during_check(&mut self) {
        use crate::types::TorrentCommand;

        while let Ok(cmd) = self.cmd_rx.try_recv() {
            match cmd {
                TorrentCommand::Stats { reply } => {
                    let _ = reply.send(self.make_stats());
                }
                TorrentCommand::ClearSaveResumeFlag => {
                    self.need_save_resume = false;
                }
                // Other commands: attempt to re-enqueue for later.
                // If the channel is full, the sender will retry.
                other => {
                    // We can't easily re-enqueue since we only have the
                    // receiver. Log and drop non-critical commands during
                    // recheck. Stats is the only command the poll loop needs.
                    tracing::debug!("dropping command during recheck: {other:?}");
                }
            }
        }
    }

    /// Fire `FileCompleted` alerts for all files whose pieces are fully verified.
    ///
    /// Used after initial check or force-recheck to emit alerts for complete files.
    pub(crate) fn fire_file_completed_alerts(&self) {
        let Some(meta) = self.meta.as_ref() else {
            return;
        };
        let Some(lengths) = self.lengths.as_ref() else {
            return;
        };
        let bitfield = match self.chunk_tracker.as_ref() {
            Some(ct) => ct.bitfield(),
            None => return,
        };

        let files = meta.info.files();
        let piece_length = lengths.piece_length();
        let mut file_offset = 0u64;

        for (file_idx, file_entry) in files.iter().enumerate() {
            let file_end = file_offset + file_entry.length;
            if file_entry.length == 0 {
                file_offset = file_end;
                continue;
            }

            let first_piece = (file_offset / piece_length) as u32;
            let last_piece = ((file_end - 1) / piece_length) as u32;

            let mut all_complete = true;
            for p in first_piece..=last_piece {
                if !bitfield.get(p) {
                    all_complete = false;
                    break;
                }
            }

            if all_complete {
                post_alert(
                    &self.alert_tx,
                    &self.alert_mask,
                    AlertKind::FileCompleted {
                        info_hash: self.info_hash,
                        file_index: file_idx,
                    },
                );
            }

            file_offset = file_end;
        }
    }

    /// Handle a force recheck request: clear all piece state, re-verify,
    /// transition to the appropriate post-check state, then send reply.
    pub(crate) async fn handle_force_recheck(
        &mut self,
        reply: tokio::sync::oneshot::Sender<crate::Result<()>>,
    ) {
        // Disconnect all peers — they hold stale bitfield state (non-blocking)
        for peer in self.peers.values() {
            let _ = peer.cmd_tx.try_send(PeerCommand::Shutdown);
        }
        self.peers.clear();

        // Clear all piece completion state
        if let Some(ref mut ct) = self.chunk_tracker {
            ct.clear();
        }

        // Transition to Checking and reply immediately so the GUI sees the
        // state change. The actual verification runs below and updates
        // checking_progress as it goes.
        self.transition_state(TorrentState::Checking);
        self.checking_progress = 0.0;
        let _ = reply.send(Ok(()));

        // Run the full verification pipeline, draining read-only commands
        // (Stats, Info, etc.) between pieces so the poll loop can observe
        // Checking state and progress.
        self.verify_existing_pieces_interactive().await;

        // M93: Rebuild atomic states after recheck
        if let Some(ct) = &self.chunk_tracker {
            let atomic_states = Arc::new(AtomicPieceStates::new(
                self.num_pieces,
                ct.bitfield(),
                &self.wanted_pieces,
            ));
            self.atomic_states = Some(Arc::clone(&atomic_states));
            self.piece_owner = vec![None; self.num_pieces as usize];
            // M103: Rebuild block stealing state after recheck
            if self.config.use_block_stealing {
                if let Some(ref lengths) = self.lengths {
                    self.block_maps = Some(Arc::new(BlockMaps::new(self.num_pieces, lengths)));
                }
                self.steal_candidates = Some(Arc::new(StealCandidates::new()));
            }
            // M120: Rebuild per-piece write guards
            self.piece_write_guards = Some(Arc::new(
                crate::piece_reservation::PieceWriteGuards::new(self.num_pieces),
            ));
        }

        // M187: Rebuild direct-acquire dispatch state (outside borrow scope).
        if let Some(ref ct) = self.chunk_tracker {
            let we_have = ct.bitfield().clone();
            self.piece_tracker = Some(crate::piece_reservation::PieceTracker::new(
                self.num_pieces,
                &we_have,
                &self.wanted_pieces,
            ));
        }
        if let Some(ref cached) = self.cached_files {
            let file_piece_ranges: Vec<(u32, u32)> = cached
                .entries
                .iter()
                .map(|e| (e.first_piece, e.last_piece))
                .collect();
            let om = Arc::new(crate::piece_reservation::PieceOrderMap::build(
                &self.file_priorities,
                &file_piece_ranges,
                self.num_pieces,
                self.order_map_tx.borrow().generation + 1,
            ));
            self.order_map_tx.send_replace(om);
        }
    }

    pub(crate) fn build_resume_data(&self) -> crate::Result<irontide_core::FastResumeData> {
        let pieces_bytes = match &self.chunk_tracker {
            Some(ct) => ct.bitfield().as_bytes().to_vec(),
            None => Vec::new(),
        };

        let name = self
            .meta
            .as_ref()
            .map(|m| m.info.name.clone())
            .unwrap_or_default();

        let save_path = self.config.download_dir.to_string_lossy().into_owned();

        let mut rd =
            irontide_core::FastResumeData::new(self.info_hash.as_bytes().to_vec(), name, save_path);

        rd.pieces = pieces_bytes;

        rd.total_uploaded = self.uploaded as i64;
        rd.total_downloaded = self.downloaded as i64;

        rd.paused = i64::from(self.state == TorrentState::Paused);
        rd.queued = i64::from(self.state == TorrentState::Queued);
        rd.seed_mode = i64::from(self.state == TorrentState::Seeding);
        rd.super_seeding = i64::from(self.super_seed.is_some());

        // Collect tracker URLs from torrent metadata and re-serialize info dict
        if let Some(ref meta) = self.meta {
            if let Some(ref announce_list) = meta.announce_list {
                rd.trackers.clone_from(announce_list);
            } else if let Some(ref announce) = meta.announce {
                rd.trackers = vec![vec![announce.clone()]];
            }
            rd.url_seeds.clone_from(&meta.url_list);
            rd.http_seeds.clone_from(&meta.httpseeds);

            // Embed the bencoded info dict so the torrent can be reconstructed
            // from resume data without the original .torrent file.
            rd.info = Some(
                irontide_bencode::to_bytes(&meta.info).map_err(irontide_core::Error::Bencode)?,
            );
        }

        // BEP 52: store v2 info hash if present (hybrid or v2-only)
        rd.info_hash2 = self.info_hashes.v2.map(|h| h.as_bytes().to_vec());

        // Timestamps
        rd.added_time = self.added_time;
        rd.completed_time = self.completed_time;
        rd.last_download = self.last_download;
        rd.last_upload = self.last_upload;

        // Durations — include the current active stint (mirrors make_stats logic)
        rd.active_time = self.active_duration
            + self
                .active_since
                .map_or(0, |since| since.elapsed().as_secs() as i64);
        rd.finished_time = self.finished_duration
            + self
                .state_duration_since
                .filter(|_| matches!(self.state, TorrentState::Complete | TorrentState::Seeding))
                .map_or(0, |since| since.elapsed().as_secs() as i64);
        rd.seeding_time = self.seeding_duration
            + self
                .state_duration_since
                .filter(|_| self.state == TorrentState::Seeding)
                .map_or(0, |since| since.elapsed().as_secs() as i64);

        // Collect connected peer addresses as compact bytes
        let peer_addrs: Vec<std::net::SocketAddr> = self.peers.keys().copied().collect();
        rd.peers = irontide_tracker::compact::encode_compact_peers(&peer_addrs);
        rd.peers6 = irontide_tracker::compact::encode_compact_peers6(&peer_addrs);

        // Per-file priorities
        rd.file_priority = self
            .file_priorities
            .iter()
            .map(|&p| i64::from(p as u8))
            .collect();

        // M170: persist category label + creator/creation_date so the
        // torrent retains its qBt-compat metadata across restart. The
        // latter two are pulled live from self.meta when available; for
        // magnet-added torrents pre-resolution they stay None and
        // backfill on the first save after metadata arrives.
        rd.category.clone_from(&self.config.category);
        if let Some(ref meta) = self.meta {
            rd.created_by.clone_from(&meta.created_by);
            rd.creation_date = meta.creation_date;
        }

        // M171: persist the qBt-compat tag list. `Vec<String>` serializes
        // as a bencode list of byte-strings; `skip_serializing_if =
        // "Vec::is_empty"` on `FastResumeData.tags` keeps older resume
        // files bit-identical on save when no tags are set. Load-side
        // plumbing lives in `SessionActor::handle_load_resume_state`,
        // which threads `rd.tags` into the new `TorrentConfig` via
        // `handle_add_torrent` / `handle_add_magnet` before the actor
        // spawns — matching the `AddTorrentParams::with_tags` semantics.
        rd.tags.clone_from(&self.config.tags);

        // M178: persist per-URL web seed stats so downloaded-byte counters
        // and last-error / consecutive-failure state survive app restart
        // (Tension-1). `skip_serializing_if = "HashMap::is_empty"` on
        // `FastResumeData.web_seed_stats` keeps older resume files (which
        // have no `web_seed_stats` key) bit-identical when no stats exist.
        rd.web_seed_stats.clone_from(&self.web_seed_stats);

        Ok(rd)
    }

    pub(crate) fn handle_set_file_priority(
        &mut self,
        index: usize,
        priority: FilePriority,
    ) -> crate::Result<()> {
        if index >= self.file_priorities.len() {
            return Err(crate::Error::InvalidFileIndex {
                index,
                count: self.file_priorities.len(),
            });
        }

        self.file_priorities[index] = priority;

        // Rebuild wanted_pieces bitfield
        if let Some(ref meta) = self.meta {
            let file_lengths: Vec<u64> = meta.info.files().iter().map(|f| f.length).collect();
            if let Some(ref lengths) = self.lengths {
                self.wanted_pieces = crate::piece_selector::build_wanted_pieces(
                    &self.file_priorities,
                    &file_lengths,
                    lengths,
                );
            }
        }

        Ok(())
    }

    pub(crate) fn sync_piece_states_with_wanted(&self) {
        let Some(ref atomic_states) = self.atomic_states else {
            return;
        };
        for piece in 0..self.num_pieces {
            let wanted = self.wanted_pieces.get(piece);
            let current = atomic_states.get(piece);
            if !wanted && current == crate::piece_reservation::PieceState::Available {
                atomic_states.mark_unwanted(piece);
            } else if wanted && current == crate::piece_reservation::PieceState::Unwanted {
                atomic_states.mark_available(piece);
            }
        }
    }

    pub(crate) fn check_seed_ratio(&mut self) -> bool {
        if self.state != TorrentState::Seeding {
            return false;
        }
        if let Some(limit) = self.config.seed_ratio_limit
            && self.downloaded > 0
        {
            let ratio = self.uploaded as f64 / self.downloaded as f64;
            if ratio >= limit {
                info!(ratio, limit, "seed ratio reached, stopping");
                self.transition_state(TorrentState::Stopped);
                return true;
            }
        }
        false
    }

    /// M171: Check whether a seed-time-based limit has been reached and, if so,
    /// stop the torrent. Mirrors `check_seed_ratio` in shape. Returns `true`
    /// when the transition to `Stopped` was triggered so the caller can
    /// shut down peers in the same tick.
    ///
    /// Two independent limits are checked (either trips the transition):
    ///
    /// - `seed_time_limit_secs` — cumulative time spent in the Seeding state
    ///   accumulated via `self.seeding_duration` (updated by `transition_state`).
    /// - `inactive_seed_time_limit_secs` — seconds elapsed since the last
    ///   outgoing Piece message while currently in Seeding.
    pub(crate) fn check_seed_time(&mut self) -> bool {
        if self.state != TorrentState::Seeding {
            return false;
        }

        // Include time spent in the current Seeding span, since
        // `self.seeding_duration` is only accumulated on state transition
        // (see `transition_state` — the duration of the state we are still
        // in has not been added yet).
        let mut active_seeding = self.seeding_duration;
        if let Some(since) = self.state_duration_since {
            let elapsed = since.elapsed().as_secs() as i64;
            active_seeding = active_seeding.saturating_add(elapsed);
        }

        if let Some(reason) = should_stop_by_seed_time(
            active_seeding,
            self.config.seed_time_limit_secs,
            self.last_upload,
            now_unix(),
            self.config.inactive_seed_time_limit_secs,
        ) {
            match reason {
                SeedTimeStopReason::Cumulative { secs, limit } => {
                    info!(
                        seeding_secs = secs,
                        limit, "seed time limit reached, stopping"
                    );
                }
                SeedTimeStopReason::Inactive { secs, limit } => {
                    info!(
                        idle_secs = secs,
                        limit, "inactive seed time limit reached, stopping"
                    );
                }
            }
            self.transition_state(TorrentState::Stopped);
            return true;
        }

        false
    }

    /// Restore a piece bitmap from resume data (M161 Phase 4).
    ///
    /// Validates the bitfield length against `num_pieces`, then replaces the
    /// chunk tracker's bitfield with the deserialized one. If the chunk
    /// tracker is not yet initialized (magnet still resolving), returns an error.
    pub(crate) fn handle_restore_resume_bitmap(&mut self, pieces: Vec<u8>) -> crate::Result<()> {
        let Some(ref lengths) = self.lengths else {
            return Err(crate::Error::InvalidSettings(
                "cannot restore bitmap: chunk tracker not initialized".into(),
            ));
        };

        if !crate::persistence::validate_resume_bitfield(&pieces, self.num_pieces) {
            return Err(crate::Error::InvalidSettings(format!(
                "resume bitmap length mismatch: got {} bytes, expected {} for {} pieces",
                pieces.len(),
                self.num_pieces.div_ceil(8),
                self.num_pieces,
            )));
        }

        let bitfield = irontide_storage::Bitfield::from_bytes(pieces, self.num_pieces)
            .map_err(|e| crate::Error::InvalidSettings(format!("invalid resume bitfield: {e}")))?;

        let new_ct = irontide_storage::ChunkTracker::from_bitfield(bitfield, lengths.clone());
        self.chunk_tracker = Some(new_ct);

        info!(
            num_pieces = self.num_pieces,
            "restored piece bitmap from resume data"
        );
        Ok(())
    }

    pub(crate) fn handle_update_settings(&mut self, delta: &crate::types::SettingsDelta) {
        if let Some(v) = delta.enable_dht {
            self.config.enable_dht = v;
        }
        if let Some(v) = delta.enable_pex {
            self.config.enable_pex = v;
        }
        if let Some(v) = delta.max_peers {
            self.config.max_peers = v;
        }
        if let Some(ref v) = delta.seed_ratio_limit {
            self.config.seed_ratio_limit = *v;
        }
        if let Some(ref v) = delta.seed_time_limit_secs {
            self.config.seed_time_limit_secs = *v;
        }
        if let Some(ref v) = delta.inactive_seed_time_limit_secs {
            self.config.inactive_seed_time_limit_secs = *v;
        }
        if let Some(v) = delta.encryption_mode {
            self.config.encryption_mode = v;
        }
        if let Some(v) = delta.anonymous_mode {
            self.config.anonymous_mode = v;
            if v {
                self.config.enable_dht = false;
            }
        }
        // M224: live per-torrent unchoke-slot cap. `n >= 1` caps the choker's
        // regular unchoke set at `n`. `-1` (unlimited) falls back to the
        // historical default of 4 regular slots so removing the cap actually
        // releases the restriction rather than freezing the prior value.
        if let Some(v) = delta.max_uploads_per_torrent {
            self.config.max_uploads_per_torrent = v;
            let slots = if v >= 1 { v as usize } else { 4 };
            self.choker.set_unchoke_slots(slots);
        }
        if let Some(v) = delta.hashing_threads {
            self.config.hashing_threads = v;
        }
    }
}

/// M171: Reason a torrent should be stopped for a seed-time-based limit.
///
/// Decoupled from `TorrentActor` so it can be unit tested without a full
/// actor construction. Used by `check_seed_time`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SeedTimeStopReason {
    /// Cumulative seeding duration reached `seed_time_limit_secs`.
    Cumulative { secs: i64, limit: u64 },
    /// Idle time since the last outgoing Piece reached
    /// `inactive_seed_time_limit_secs`.
    Inactive { secs: i64, limit: u64 },
}

/// M171: Pure helper deciding whether a seed-time limit has been crossed.
///
/// Returns `Some(reason)` when the torrent should transition to `Stopped`,
/// `None` otherwise. Preconditions (caller enforces):
/// * The torrent is currently in the `Seeding` state.
///
/// Semantics:
/// * Cumulative limit trips when `active_seeding_secs >= limit` (>= 0).
/// * Inactive limit trips when `last_upload > 0` (i.e. at least one Piece
///   has been served) and `now_unix - last_upload >= limit`. A never-active
///   seeder never trips the inactive branch — that is qBt parity.
pub(crate) fn should_stop_by_seed_time(
    active_seeding_secs: i64,
    seed_time_limit_secs: Option<u64>,
    last_upload_unix: i64,
    now_unix_secs: i64,
    inactive_seed_time_limit_secs: Option<u64>,
) -> Option<SeedTimeStopReason> {
    if let Some(limit) = seed_time_limit_secs
        && active_seeding_secs >= 0
        && (active_seeding_secs as u64) >= limit
    {
        return Some(SeedTimeStopReason::Cumulative {
            secs: active_seeding_secs,
            limit,
        });
    }

    if let Some(limit) = inactive_seed_time_limit_secs
        && last_upload_unix > 0
    {
        let idle = now_unix_secs.saturating_sub(last_upload_unix);
        if idle >= 0 && (idle as u64) >= limit {
            return Some(SeedTimeStopReason::Inactive { secs: idle, limit });
        }
    }

    None
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn seed_time_no_limits_never_stops() {
        assert_eq!(
            should_stop_by_seed_time(10_000, None, 0, 20_000, None),
            None
        );
        assert_eq!(
            should_stop_by_seed_time(10_000, None, 19_999, 20_000, None),
            None
        );
    }

    #[test]
    fn seed_time_cumulative_under_limit_continues() {
        assert_eq!(
            should_stop_by_seed_time(100, Some(3600), 0, 0, None),
            None,
            "100s < 3600s limit — should not stop"
        );
    }

    #[test]
    fn seed_time_cumulative_over_limit_stops() {
        let reason = should_stop_by_seed_time(3600, Some(3600), 0, 0, None);
        assert_eq!(
            reason,
            Some(SeedTimeStopReason::Cumulative {
                secs: 3600,
                limit: 3600,
            })
        );
    }

    #[test]
    fn seed_time_cumulative_well_over_limit_stops() {
        let reason = should_stop_by_seed_time(10_000, Some(3600), 0, 0, None);
        assert!(matches!(
            reason,
            Some(SeedTimeStopReason::Cumulative { .. })
        ));
    }

    #[test]
    fn seed_time_inactive_under_limit_continues() {
        // idle = 20000 - 19100 = 900 < 1800 limit
        assert_eq!(
            should_stop_by_seed_time(100, None, 19_100, 20_000, Some(1800)),
            None
        );
    }

    #[test]
    fn seed_time_inactive_over_limit_stops() {
        // idle = 20000 - 10000 = 10000 > 3600 limit
        let reason = should_stop_by_seed_time(100, None, 10_000, 20_000, Some(3600));
        assert!(matches!(reason, Some(SeedTimeStopReason::Inactive { .. })));
    }

    #[test]
    fn seed_time_inactive_requires_prior_upload() {
        // last_upload = 0 means we have never uploaded — do not trip idle.
        assert_eq!(
            should_stop_by_seed_time(100, None, 0, 20_000, Some(60)),
            None,
            "never-active seeder must not trip inactive limit"
        );
    }

    #[test]
    fn seed_time_cumulative_wins_over_inactive() {
        // Both limits tripped — cumulative is checked first and returned.
        let reason = should_stop_by_seed_time(10_000, Some(3600), 100, 20_000, Some(60));
        assert!(matches!(
            reason,
            Some(SeedTimeStopReason::Cumulative { .. })
        ));
    }

    #[test]
    fn seed_time_negative_active_safe() {
        // Defensive: even with a pathological negative duration, we don't stop.
        assert_eq!(
            should_stop_by_seed_time(-1, Some(3600), 0, 0, None),
            None,
            "negative cumulative duration must not trip limit"
        );
    }

    #[test]
    fn torrent_stats_default_user_forced_false() {
        let stats = crate::types::TorrentStats::default();
        assert!(!stats.user_forced);
        assert!(stats.seed_ratio_override.is_none());
    }

    #[test]
    fn torrent_summary_carries_user_forced() {
        let stats = crate::types::TorrentStats {
            user_forced: true,
            ..Default::default()
        };
        let summary = crate::types::TorrentSummary::from(&stats);
        assert!(summary.user_forced);
    }

    #[test]
    fn torrent_stats_seed_ratio_override_serializes() {
        let stats = crate::types::TorrentStats {
            seed_ratio_override: Some(2.5),
            ..Default::default()
        };
        let json = serde_json::to_string(&stats).unwrap();
        assert!(json.contains("\"seed_ratio_override\":2.5"));

        let stats_none = crate::types::TorrentStats::default();
        let json_none = serde_json::to_string(&stats_none).unwrap();
        assert!(!json_none.contains("seed_ratio_override"));
    }
}