use std::collections::VecDeque;

use crate::current_repo_index::{CurrentRepoIndex, CurrentRepoIndexHandle};
use crate::parallel::config::ParallelPhase1Config;
use crate::parallel::dead_repo_blacklist::{
    DeadRepoBlacklist, DeadRepoFailureOutcome, DeadRepoSuccessOutcome,
};
use crate::parallel::repo_scheduler::{
    InFlightRepoTable, RepoCompletion, RepoRequestAction, TransportCompletion,
    TransportRequestAction, TransportStateTables,
};
use crate::parallel::stats::ParallelRunStats;
use crate::parallel::types::{
    RepoDedupKey, RepoIdentity, RepoKey, RepoRequester, RepoSyncResultEnvelope, RepoSyncTask,
    RepoTransportErrorClass, RepoTransportMode, RepoTransportResultEnvelope,
    RepoTransportResultKind, RepoTransportTask, TalInputSpec,
};
use crate::policy::SyncPreference;

pub struct GlobalRunCoordinator {
    pub config: ParallelPhase1Config,
    pub tal_inputs: Vec<TalInputSpec>,
    pub current_repo_index: CurrentRepoIndexHandle,
    pub in_flight_repos: InFlightRepoTable,
    pub transport_tables: TransportStateTables,
    pub pending_repo_tasks: VecDeque<RepoSyncTask>,
    pub pending_transport_tasks: VecDeque<RepoTransportTask>,
    pub stats: ParallelRunStats,
    /// Mutable working copy of the dead-repo blacklist (#141): failure/success
    /// counters update during the run and persist at run end. Skip decisions
    /// inside `transport_tables` use the frozen run-start snapshot.
    pub dead_repo_blacklist: Option<DeadRepoBlacklist>,
}

impl GlobalRunCoordinator {
    pub fn new(config: ParallelPhase1Config, tal_inputs: Vec<TalInputSpec>) -> Self {
        let dead_repo_blacklist = config
            .dead_repo_blacklist
            .as_ref()
            .map(|blacklist_config| {
                let (blacklist, warning) = DeadRepoBlacklist::load(&blacklist_config.path);
                if let Some(warning) = warning {
                    crate::progress_log::emit(
                        "dead_repo_blacklist_load_warning",
                        serde_json::json!({ "warning": warning }),
                    );
                }
                blacklist
            });
        let mut transport_tables = TransportStateTables::new();
        if let Some(blacklist) = dead_repo_blacklist.as_ref() {
            // Frozen run-start snapshot for skip decisions.
            transport_tables.set_dead_repo_blacklist(blacklist.clone());
        }
        Self {
            config,
            tal_inputs,
            current_repo_index: CurrentRepoIndex::shared(),
            in_flight_repos: InFlightRepoTable::new(),
            transport_tables,
            pending_repo_tasks: VecDeque::new(),
            pending_transport_tasks: VecDeque::new(),
            stats: ParallelRunStats::default(),
            dead_repo_blacklist,
        }
    }

    pub fn current_repo_index_handle(&self) -> CurrentRepoIndexHandle {
        self.current_repo_index.clone()
    }

    pub fn register_repo_request(
        &mut self,
        repo_key: RepoKey,
        requester: RepoRequester,
        validation_time: time::OffsetDateTime,
        sync_preference: SyncPreference,
        priority: u8,
    ) -> RepoRequestAction {
        let action = self.in_flight_repos.register_request(
            repo_key,
            requester,
            validation_time,
            sync_preference,
            priority,
        );

        match &action {
            RepoRequestAction::Enqueued(task) => {
                self.stats.repo_tasks_total += 1;
                self.pending_repo_tasks.push_back(task.clone());
                self.stats.repo_queue_depth = self.pending_repo_tasks.len();
            }
            RepoRequestAction::Reused(_) | RepoRequestAction::FailedReuse { .. } => {
                self.stats.repo_tasks_reused += 1;
            }
            RepoRequestAction::Waiting => {}
        }

        action
    }

    pub fn pop_next_repo_task(&mut self) -> Option<RepoSyncTask> {
        let next = self.pending_repo_tasks.pop_front();
        self.stats.repo_queue_depth = self.pending_repo_tasks.len();
        next
    }

    pub fn mark_repo_running(
        &mut self,
        repo_key: &RepoKey,
        started_at: time::OffsetDateTime,
    ) -> Result<(), String> {
        self.in_flight_repos.mark_running(repo_key, started_at)?;
        self.stats.repo_tasks_running += 1;
        Ok(())
    }

    pub fn complete_repo_success(
        &mut self,
        result: RepoSyncResultEnvelope,
        finished_at: time::OffsetDateTime,
    ) -> Result<RepoCompletion, String> {
        let repo_key = result.repo_key.clone();
        let completion = self
            .in_flight_repos
            .complete_success(&repo_key, result, finished_at)?;
        self.stats.repo_tasks_running = self.stats.repo_tasks_running.saturating_sub(1);
        Ok(completion)
    }

    pub fn complete_repo_failure(
        &mut self,
        result: RepoSyncResultEnvelope,
        finished_at: time::OffsetDateTime,
    ) -> Result<RepoCompletion, String> {
        let repo_key = result.repo_key.clone();
        let completion = self
            .in_flight_repos
            .complete_failure(&repo_key, result, finished_at)?;
        self.stats.repo_tasks_running = self.stats.repo_tasks_running.saturating_sub(1);
        self.stats.repo_tasks_failed += 1;
        Ok(completion)
    }

    pub fn register_transport_request(
        &mut self,
        identity: RepoIdentity,
        requester: RepoRequester,
        validation_time: time::OffsetDateTime,
        priority: u8,
        rsync_scope_uri: String,
        rsync_failure_scope_uri: Option<String>,
        sync_preference: SyncPreference,
        retry_short_timeout: bool,
    ) -> TransportRequestAction {
        let mut action = self.transport_tables.register_transport_request(
            identity,
            requester,
            validation_time,
            priority,
            rsync_scope_uri,
            rsync_failure_scope_uri,
            sync_preference,
        );
        match &mut action {
            TransportRequestAction::Enqueue(task) => {
                if retry_short_timeout {
                    task.retry_short_timeout = true;
                }
                self.stats.repo_tasks_total += 1;
                self.pending_transport_tasks.push_back(task.clone());
                self.stats.repo_queue_depth = self.pending_transport_tasks.len();
            }
            TransportRequestAction::ReusedSuccess(_)
            | TransportRequestAction::ReusedTerminalFailure(_) => {
                self.stats.repo_tasks_reused += 1;
            }
            TransportRequestAction::Waiting { .. } => {}
        }
        action
    }

    pub fn push_transport_task(&mut self, task: RepoTransportTask) {
        self.stats.repo_tasks_total += 1;
        self.pending_transport_tasks.push_back(task);
        self.stats.repo_queue_depth = self.pending_transport_tasks.len();
    }

    pub fn pop_next_transport_task(&mut self) -> Option<RepoTransportTask> {
        let next = self.pending_transport_tasks.pop_front();
        self.stats.repo_queue_depth = self.pending_transport_tasks.len();
        next
    }

    pub fn mark_transport_running(
        &mut self,
        dedup_key: &crate::parallel::types::RepoDedupKey,
        started_at: time::OffsetDateTime,
    ) -> Result<(), String> {
        self.transport_tables
            .mark_transport_running(dedup_key, started_at)?;
        self.stats.repo_tasks_running += 1;
        Ok(())
    }

    pub fn complete_transport_result(
        &mut self,
        result: RepoTransportResultEnvelope,
        finished_at: time::OffsetDateTime,
    ) -> Result<TransportCompletion, String> {
        let completion = self
            .transport_tables
            .complete_transport_result(result.clone(), finished_at)?;
        self.stats.repo_tasks_running = self.stats.repo_tasks_running.saturating_sub(1);
        if matches!(
            result.result,
            crate::parallel::types::RepoTransportResultKind::Failed { .. }
        ) && result.mode == crate::parallel::types::RepoTransportMode::Rsync
        {
            self.stats.repo_tasks_failed += 1;
        }
        self.update_dead_repo_blacklist(&result, finished_at);
        Ok(completion)
    }

    /// Update dead-repo blacklist counters (#141) from a real worker result.
    /// Runs on the single pump thread; synthesized terminal envelopes never
    /// pass through here, so no double counting is possible.
    fn update_dead_repo_blacklist(
        &mut self,
        result: &RepoTransportResultEnvelope,
        finished_at: time::OffsetDateTime,
    ) {
        let Some(blacklist_config) = self.config.dead_repo_blacklist.clone() else {
            return;
        };
        let Some(blacklist) = self.dead_repo_blacklist.as_mut() else {
            return;
        };
        let now_unix = finished_at.unix_timestamp().max(0) as u64;
        let key: Option<(RepoTransportMode, String)> = match (&result.dedup_key, &result.result) {
            (RepoDedupKey::RrdpNotify { notification_uri }, RepoTransportResultKind::Success { .. }) => {
                Some((RepoTransportMode::Rrdp, notification_uri.clone()))
            }
            (RepoDedupKey::RsyncScope { .. }, RepoTransportResultKind::Success { .. }) => {
                Some((RepoTransportMode::Rsync, result.repo_identity.rsync_base_uri.clone()))
            }
            (
                RepoDedupKey::RrdpNotify { notification_uri },
                RepoTransportResultKind::Failed {
                    error_class: RepoTransportErrorClass::TransportFetch,
                    ..
                },
            ) => Some((RepoTransportMode::Rrdp, notification_uri.clone())),
            (
                RepoDedupKey::RsyncScope { .. },
                RepoTransportResultKind::Failed {
                    error_class: RepoTransportErrorClass::TransportFetch,
                    ..
                },
            ) => Some((RepoTransportMode::Rsync, result.repo_identity.rsync_base_uri.clone())),
            _ => None,
        };
        let Some((transport, uri)) = key else {
            return;
        };
        match &result.result {
            RepoTransportResultKind::Success { .. } => {
                if let Some(outcome) = blacklist.record_transport_success(transport, &uri) {
                    crate::progress_log::emit(
                        "dead_repo_blacklist_remove",
                        serde_json::json!({
                            "transport": transport.as_str(),
                            "uri": uri,
                            "reason": match outcome {
                                DeadRepoSuccessOutcome::CounterReset => "counter_reset",
                                DeadRepoSuccessOutcome::BlacklistRemoved => "fetch_success",
                            },
                        }),
                    );
                }
            }
            RepoTransportResultKind::Failed { .. } => {
                if let DeadRepoFailureOutcome::Admitted = blacklist.record_transport_failure(
                    transport,
                    &uri,
                    now_unix,
                    blacklist_config.fail_threshold,
                    blacklist_config.capacity,
                ) {
                    crate::progress_log::emit(
                        "dead_repo_blacklist_admit",
                        serde_json::json!({
                            "transport": transport.as_str(),
                            "uri": uri,
                            "fail_threshold": blacklist_config.fail_threshold,
                        }),
                    );
                }
            }
        }
    }

    pub fn dead_repo_blacklist(&self) -> Option<&DeadRepoBlacklist> {
        self.dead_repo_blacklist.as_ref()
    }

    pub fn runtime_record(
        &self,
        identity: &RepoIdentity,
    ) -> Option<&crate::parallel::repo_scheduler::RepoRuntimeRecord> {
        self.transport_tables.runtime_record(identity)
    }

    pub fn finalized_runtime_records_for_transport(
        &self,
        dedup_key: &RepoDedupKey,
    ) -> Vec<crate::parallel::repo_scheduler::RepoRuntimeRecord> {
        self.transport_tables
            .finalized_runtime_records_for_transport(dedup_key)
    }

    pub fn finalized_runtime_records_for_transport_result(
        &self,
        result: &RepoTransportResultEnvelope,
    ) -> Vec<crate::parallel::repo_scheduler::RepoRuntimeRecord> {
        self.transport_tables
            .finalized_runtime_records_for_transport_result(result)
    }

    pub fn reset_run_state(&mut self) {
        self.in_flight_repos.reset_run_state();
        self.transport_tables.reset_run_state();
        self.pending_repo_tasks.clear();
        self.pending_transport_tasks.clear();
        self.stats = ParallelRunStats::default();
        if let Ok(mut index) = self.current_repo_index.write() {
            index.clear();
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::parallel::config::ParallelPhase1Config;
    use crate::parallel::repo_scheduler::RepoRequestAction;
    use crate::parallel::run_coordinator::GlobalRunCoordinator;
    use crate::parallel::types::{
        RepoIdentity, RepoKey, RepoRequester, RepoSyncResultEnvelope, RepoSyncResultKind,
        RepoSyncResultRef, TalInputSpec,
    };
    use crate::policy::SyncPreference;
    use crate::storage::{RepositoryViewEntry, RepositoryViewState};

    fn requester(tal_id: &str, rir_id: &str, manifest: &str) -> RepoRequester {
        RepoRequester {
            tal_id: tal_id.to_string(),
            rir_id: rir_id.to_string(),
            parent_node_id: None,
            ca_instance_handle_id: format!("{tal_id}:{manifest}"),
            publication_point_rsync_uri: "rsync://example.test/repo/".to_string(),
            manifest_rsync_uri: manifest.to_string(),
        }
    }

    #[test]
    fn coordinator_holds_tal_inputs_and_default_stats() {
        let coordinator = GlobalRunCoordinator::new(
            ParallelPhase1Config::default(),
            vec![TalInputSpec::from_url("https://example.test/arin.tal")],
        );
        assert_eq!(coordinator.tal_inputs.len(), 1);
        assert_eq!(coordinator.tal_inputs[0].tal_id, "arin");
        assert_eq!(coordinator.stats.repo_tasks_total, 0);
        assert!(coordinator.in_flight_repos.is_empty());
    }

    #[test]
    fn coordinator_enqueues_new_repo_task_once() {
        let mut coordinator = GlobalRunCoordinator::new(
            ParallelPhase1Config::default(),
            vec![TalInputSpec::from_url("https://example.test/arin.tal")],
        );
        let key = RepoKey::new("rsync://example.test/repo/", None);
        let action = coordinator.register_repo_request(
            key.clone(),
            requester("arin", "arin", "rsync://example.test/repo/root.mft"),
            time::OffsetDateTime::UNIX_EPOCH,
            SyncPreference::RrdpThenRsync,
            7,
        );
        let task = match action {
            RepoRequestAction::Enqueued(task) => task,
            other => panic!("expected enqueue, got {other:?}"),
        };
        assert_eq!(task.priority, 7);
        assert_eq!(coordinator.stats.repo_tasks_total, 1);
        assert_eq!(coordinator.stats.repo_queue_depth, 1);

        let popped = coordinator.pop_next_repo_task().expect("task queued");
        assert_eq!(popped.repo_key, key);
        assert_eq!(coordinator.stats.repo_queue_depth, 0);
    }

    #[test]
    fn coordinator_reset_run_state_clears_runtime_only_state() {
        let mut coordinator = GlobalRunCoordinator::new(
            ParallelPhase1Config::default(),
            vec![TalInputSpec::from_url("https://example.test/arin.tal")],
        );
        let identity = RepoIdentity::new(
            Some("https://example.test/notify.xml".to_string()),
            "rsync://example.test/repo/",
        );
        let requester = requester("arin", "arin", "rsync://example.test/repo/root.mft");
        let action = coordinator.register_transport_request(
            identity.clone(),
            requester,
            time::OffsetDateTime::UNIX_EPOCH,
            0,
            "rsync://example.test/repo/".to_string(),
            None,
            SyncPreference::RrdpThenRsync,
            false,
        );
        assert!(matches!(
            action,
            crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(_)
        ));
        assert!(coordinator.runtime_record(&identity).is_some());
        assert_eq!(coordinator.pending_transport_tasks.len(), 1);

        {
            let mut index = coordinator.current_repo_index.write().expect("index write lock");
            index
                .apply_repository_view_entries(&[RepositoryViewEntry {
                    rsync_uri: "rsync://example.test/repo/a.roa".to_string(),
                    current_hash: Some("aa".repeat(32)),
                    repository_source: Some("rsync://example.test/repo/".to_string()),
                    object_type: Some("roa".to_string()),
                    state: RepositoryViewState::Present,
                }])
                .expect("apply current object");
            assert_eq!(index.active_uri_count(), 1);
        }

        coordinator.reset_run_state();

        assert!(coordinator.runtime_record(&identity).is_none());
        assert!(coordinator.pending_transport_tasks.is_empty());
        assert!(coordinator.pending_repo_tasks.is_empty());
        assert_eq!(
            coordinator
                .current_repo_index
                .read()
                .expect("index read lock")
                .active_uri_count(),
            0
        );
        assert_eq!(coordinator.stats.repo_tasks_total, 0);
    }

    #[test]
    fn coordinator_merges_waiting_requesters_and_reuses_success() {
        let mut coordinator = GlobalRunCoordinator::new(
            ParallelPhase1Config::default(),
            vec![
                TalInputSpec::from_url("https://example.test/arin.tal"),
                TalInputSpec::from_url("https://example.test/apnic.tal"),
            ],
        );
        let key = RepoKey::new("rsync://shared.example/repo/", None);
        let _ = coordinator.register_repo_request(
            key.clone(),
            requester("arin", "arin", "rsync://shared.example/repo/root.mft"),
            time::OffsetDateTime::UNIX_EPOCH,
            SyncPreference::RrdpThenRsync,
            0,
        );
        let _ = coordinator.pop_next_repo_task();
        coordinator
            .mark_repo_running(&key, time::OffsetDateTime::UNIX_EPOCH)
            .expect("running");

        let wait_action = coordinator.register_repo_request(
            key.clone(),
            requester("apnic", "apnic", "rsync://shared.example/repo/child.mft"),
            time::OffsetDateTime::UNIX_EPOCH,
            SyncPreference::RrdpThenRsync,
            0,
        );
        assert_eq!(wait_action, RepoRequestAction::Waiting);

        let completion = coordinator
            .complete_repo_success(
                RepoSyncResultEnvelope {
                    repo_key: key.clone(),
                    tal_id: "arin".to_string(),
                    rir_id: "arin".to_string(),
                    result: RepoSyncResultKind::Success(RepoSyncResultRef {
                        repo_key: key.clone(),
                        source: "rrdp".to_string(),
                    }),
                    phase: Some("rrdp_ok".to_string()),
                    timing_ms: 12,
                    warnings: Vec::new(),
                },
                time::OffsetDateTime::UNIX_EPOCH,
            )
            .expect("success");
        assert_eq!(completion.released_requesters.len(), 1);
        assert_eq!(coordinator.stats.repo_tasks_running, 0);

        let reuse_action = coordinator.register_repo_request(
            key,
            requester("ripe", "ripe", "rsync://shared.example/repo/again.mft"),
            time::OffsetDateTime::UNIX_EPOCH,
            SyncPreference::RrdpThenRsync,
            0,
        );
        assert!(matches!(reuse_action, RepoRequestAction::Reused(_)));
        assert_eq!(coordinator.stats.repo_tasks_reused, 1);
    }

    #[test]
    fn coordinator_reuses_failures_without_enqueuing_again() {
        let mut coordinator = GlobalRunCoordinator::new(
            ParallelPhase1Config::default(),
            vec![TalInputSpec::from_url("https://example.test/arin.tal")],
        );
        let key = RepoKey::new("rsync://shared.example/repo/", None);
        let _ = coordinator.register_repo_request(
            key.clone(),
            requester("arin", "arin", "rsync://shared.example/repo/root.mft"),
            time::OffsetDateTime::UNIX_EPOCH,
            SyncPreference::RrdpThenRsync,
            0,
        );
        let _ = coordinator.pop_next_repo_task();
        coordinator
            .mark_repo_running(&key, time::OffsetDateTime::UNIX_EPOCH)
            .expect("running");
        coordinator
            .complete_repo_failure(
                RepoSyncResultEnvelope {
                    repo_key: key.clone(),
                    tal_id: "arin".to_string(),
                    rir_id: "arin".to_string(),
                    result: RepoSyncResultKind::Failed {
                        detail: "timeout".to_string(),
                    },
                    phase: Some("rrdp_failed_rsync_failed".to_string()),
                    timing_ms: 12,
                    warnings: Vec::new(),
                },
                time::OffsetDateTime::UNIX_EPOCH,
            )
            .expect("failure");
        assert_eq!(coordinator.stats.repo_tasks_failed, 1);

        let action = coordinator.register_repo_request(
            key,
            requester("ripe", "ripe", "rsync://shared.example/repo/child.mft"),
            time::OffsetDateTime::UNIX_EPOCH,
            SyncPreference::RrdpThenRsync,
            0,
        );
        assert_eq!(
            action,
            RepoRequestAction::FailedReuse {
                detail: "timeout".to_string()
            }
        );
        assert_eq!(coordinator.stats.repo_tasks_total, 1);
        assert_eq!(coordinator.stats.repo_tasks_reused, 1);
    }

    // ---- Dead-repo blacklist (#141) integration tests ----

    use crate::parallel::dead_repo_blacklist::{DeadRepoBlacklist, DeadRepoBlacklistConfig};
    use crate::parallel::types::{
        RepoDedupKey, RepoTransportErrorClass, RepoTransportMode, RepoTransportResultEnvelope,
        RepoTransportResultKind,
    };

    const DEAD_NOTIFY: &str = "https://dead.example/notification.xml";
    const DEAD_BASE: &str = "rsync://dead.example/repo/";

    fn blacklist_config(fail_threshold: u32) -> DeadRepoBlacklistConfig {
        DeadRepoBlacklistConfig {
            path: std::env::temp_dir().join(format!(
                "dead_repo_blacklist_coordinator_test_{}.json",
                std::process::id()
            )),
            fail_threshold,
            capacity: 256,
        }
    }

    fn coordinator_with_blacklist(
        fail_threshold: u32,
        seed: DeadRepoBlacklist,
    ) -> GlobalRunCoordinator {
        let mut config = ParallelPhase1Config::default();
        config.dead_repo_blacklist = Some(blacklist_config(fail_threshold));
        let mut coordinator = GlobalRunCoordinator::new(
            config,
            vec![TalInputSpec::from_url("https://example.test/arin.tal")],
        );
        coordinator.dead_repo_blacklist = Some(seed.clone());
        coordinator.transport_tables.set_dead_repo_blacklist(seed);
        coordinator
    }

    fn dead_identity() -> RepoIdentity {
        RepoIdentity::new(Some(DEAD_NOTIFY.to_string()), DEAD_BASE)
    }

    fn dead_requester() -> RepoRequester {
        requester("arin", "arin", "rsync://dead.example/repo/root.mft")
    }

    fn transport_envelope(
        mode: RepoTransportMode,
        error_class: RepoTransportErrorClass,
    ) -> RepoTransportResultEnvelope {
        let (dedup_key, detail) = match mode {
            RepoTransportMode::Rrdp => (
                RepoDedupKey::RrdpNotify {
                    notification_uri: DEAD_NOTIFY.to_string(),
                },
                "http request failed: connect timeout".to_string(),
            ),
            RepoTransportMode::Rsync => (
                RepoDedupKey::RsyncScope {
                    rsync_scope_uri: DEAD_BASE.to_string(),
                },
                "rsync error: timeout waiting for daemon connection".to_string(),
            ),
        };
        RepoTransportResultEnvelope {
            dedup_key,
            rsync_failure_scope_uri: None,
            repo_identity: dead_identity(),
            mode,
            tal_id: "arin".to_string(),
            rir_id: "arin".to_string(),
            timing_ms: 1,
            result: RepoTransportResultKind::Failed {
                detail,
                warnings: Vec::new(),
                error_class,
            },
        }
    }

    fn register_dead(
        coordinator: &mut GlobalRunCoordinator,
    ) -> crate::parallel::repo_scheduler::TransportRequestAction {
        coordinator.register_transport_request(
            dead_identity(),
            dead_requester(),
            time::OffsetDateTime::UNIX_EPOCH,
            0,
            DEAD_BASE.to_string(),
            None,
            SyncPreference::RrdpThenRsync,
            false,
        )
    }

    #[test]
    fn blacklist_admission_after_threshold_transport_failures() {
        let mut coordinator = coordinator_with_blacklist(2, DeadRepoBlacklist::new());
        for _ in 0..2 {
            let action = register_dead(&mut coordinator);
            assert!(matches!(
                action,
                crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(_)
            ));
            coordinator
                .complete_transport_result(
                    transport_envelope(
                        RepoTransportMode::Rrdp,
                        RepoTransportErrorClass::TransportFetch,
                    ),
                    time::OffsetDateTime::UNIX_EPOCH,
                )
                .expect("complete");
            // Simulate the next run: run-local scheduler state resets, the
            // persistent blacklist working copy carries over.
            coordinator.reset_run_state();
        }
        let blacklist = coordinator.dead_repo_blacklist().expect("blacklist");
        assert!(blacklist.is_blacklisted(RepoTransportMode::Rrdp, DEAD_NOTIFY));
        // Admission mid-run does not affect this run's frozen snapshot: the
        // scheduler still enqueues rrdp for a fresh identity registration.
        coordinator.reset_run_state();
        let action = register_dead(&mut coordinator);
        assert!(matches!(
            action,
            crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(_)
        ));
    }

    #[test]
    fn blacklist_ignores_protocol_and_unknown_failures() {
        let mut coordinator = coordinator_with_blacklist(1, DeadRepoBlacklist::new());
        for error_class in [
            RepoTransportErrorClass::Protocol,
            RepoTransportErrorClass::Unknown,
            RepoTransportErrorClass::Storage,
        ] {
            let action = register_dead(&mut coordinator);
            assert!(matches!(
                action,
                crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(_)
            ));
            coordinator
                .complete_transport_result(
                    transport_envelope(RepoTransportMode::Rrdp, error_class),
                    time::OffsetDateTime::UNIX_EPOCH,
                )
                .expect("complete");
            coordinator.reset_run_state();
        }
        assert!(
            coordinator
                .dead_repo_blacklist()
                .expect("blacklist")
                .is_empty()
        );
    }

    #[test]
    fn blacklist_skip_rrdp_routes_straight_to_rsync() {
        let mut seed = DeadRepoBlacklist::new();
        seed.record_transport_failure(RepoTransportMode::Rrdp, DEAD_NOTIFY, 100, 1, 256);
        let mut coordinator = coordinator_with_blacklist(3, seed);
        let action = register_dead(&mut coordinator);
        let task = match action {
            crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(task) => task,
            other => panic!("expected rsync enqueue, got {other:?}"),
        };
        assert_eq!(task.mode, RepoTransportMode::Rsync);
        // register_dead passes retry_short_timeout=false (mirroring a prefetch
        // snapshot that recorded no real rrdp failure for skipped runs); the
        // skip path must still keep the short retry profile (#141).
        assert!(task.retry_short_timeout);
    }

    #[test]
    fn blacklist_dual_dead_terminates_immediately() {
        let mut seed = DeadRepoBlacklist::new();
        seed.record_transport_failure(RepoTransportMode::Rrdp, DEAD_NOTIFY, 100, 1, 256);
        seed.record_transport_failure(RepoTransportMode::Rsync, DEAD_BASE, 100, 1, 256);
        let mut coordinator = coordinator_with_blacklist(3, seed);
        let action = register_dead(&mut coordinator);
        let envelope = match action {
            crate::parallel::repo_scheduler::TransportRequestAction::ReusedTerminalFailure(
                envelope,
            ) => envelope,
            other => panic!("expected terminal failure, got {other:?}"),
        };
        match &envelope.result {
            RepoTransportResultKind::Failed {
                detail, warnings, ..
            } => {
                assert!(detail.contains("dead repo blacklist"));
                assert!(warnings
                    .iter()
                    .any(|warning| warning.message.contains("dead_repo_blacklist_skip_all")));
            }
            other => panic!("expected failed result, got {other:?}"),
        }
        let record = coordinator
            .runtime_record(&dead_identity())
            .expect("record");
        assert_eq!(
            record.state,
            crate::parallel::types::RepoRuntimeState::FailedTerminal
        );
        assert_eq!(coordinator.stats.repo_tasks_total, 0);
        assert_eq!(coordinator.stats.repo_tasks_reused, 1);
    }

    #[test]
    fn blacklist_rsync_dead_terminates_after_rrdp_failure() {
        // Only rsync blacklisted: rrdp is attempted; when it fails, the rsync
        // fallback is short-circuited into an instant terminal failure.
        let mut seed = DeadRepoBlacklist::new();
        seed.record_transport_failure(RepoTransportMode::Rsync, DEAD_BASE, 100, 1, 256);
        let mut coordinator = coordinator_with_blacklist(3, seed);
        let action = register_dead(&mut coordinator);
        assert!(matches!(
            action,
            crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(_)
        ));
        let completion = coordinator
            .complete_transport_result(
                transport_envelope(RepoTransportMode::Rrdp, RepoTransportErrorClass::TransportFetch),
                time::OffsetDateTime::UNIX_EPOCH,
            )
            .expect("complete");
        assert!(completion.follow_up_tasks.is_empty());
        let record = coordinator
            .runtime_record(&dead_identity())
            .expect("record");
        assert_eq!(
            record.state,
            crate::parallel::types::RepoRuntimeState::FailedTerminal
        );
        let terminal = record.terminal_failure.as_ref().expect("terminal failure");
        match &terminal.result {
            RepoTransportResultKind::Failed { detail, .. } => {
                assert!(detail.contains("dead repo blacklist"));
            }
            other => panic!("expected failed result, got {other:?}"),
        }
    }

    #[test]
    fn blacklist_success_resets_failure_counter() {
        let mut coordinator = coordinator_with_blacklist(3, DeadRepoBlacklist::new());
        // One transport failure starts counting.
        let action = register_dead(&mut coordinator);
        assert!(matches!(
            action,
            crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(_)
        ));
        coordinator
            .complete_transport_result(
                transport_envelope(RepoTransportMode::Rrdp, RepoTransportErrorClass::TransportFetch),
                time::OffsetDateTime::UNIX_EPOCH,
            )
            .expect("complete");
        assert_eq!(coordinator.dead_repo_blacklist().expect("blacklist").len(), 1);

        // A later success clears the entry.
        coordinator.reset_run_state();
        let action = register_dead(&mut coordinator);
        assert!(matches!(
            action,
            crate::parallel::repo_scheduler::TransportRequestAction::Enqueue(_)
        ));
        let mut success = transport_envelope(RepoTransportMode::Rrdp, RepoTransportErrorClass::Unknown);
        success.result = RepoTransportResultKind::Success {
            source: "rrdp".to_string(),
            warnings: Vec::new(),
        };
        coordinator
            .complete_transport_result(success, time::OffsetDateTime::UNIX_EPOCH)
            .expect("complete");
        assert!(
            coordinator
                .dead_repo_blacklist()
                .expect("blacklist")
                .is_empty()
        );
    }
}