//! Shared channel activity feature policy and runtime helpers.
//!
//! This module handles per-channel activity policy (typing indicators and read
//! receipts) and the runtime helpers that drive those side effects.
//!
//! Activity subsystem contract:
//! - `WsServerState` owns the runtime activity service and is the only runtime
//!   shutdown entrypoint.
//! - shutdown closes intake first, then drains already-queued work until a
//!   deadline, joining the real worker threads directly.
//! - work still queued after the deadline is dropped explicitly with logging.
//! - receive-time claims are completed immediately after durable inbound append;
//!   outbound delivery never carries read-receipt state.
//! - read receipts are committed as non-lossy in-process obligations so inbound
//!   append never waits on receipt-worker capacity.
//! - persisted read-receipt task state is versioned by the current task kind.
//! - activity-capable channel implementations must bound their own blocking
//!   I/O; the dispatcher does not spawn detached per-operation timeout threads.
//! - config reload only affects future polls/messages because each receive loop
//!   iteration snapshots its activity policy before polling and dispatch.
//! - enabled-but-unsupported features warn both at runtime and when startup or
//!   config reload evaluates channel capabilities against policy.

use std::any::Any;
use std::collections::{HashMap, HashSet};
#[cfg(test)]
use std::io;
use std::panic::{catch_unwind, resume_unwind, AssertUnwindSafe};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering};
use std::sync::mpsc as sync_mpsc;
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};

use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use thiserror::Error;
use tokio::runtime::{Handle, RuntimeFlavor};
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

use crate::plugins::{
    BindingError, ChannelPluginInstance, PluginRegistry, ReadReceiptContext, TypingContext,
};
use crate::runtime_bridge::run_sync_blocking_send;
use crate::server::ws::WsServerState;
use crate::tasks::{TaskExecutionOutcome, TaskExecutor, TaskQueue};
use crate::thread_util::{spawn_startup_named_thread_with_spawner, NamedThreadSpawner};
use crate::StartupThreadSpawnError;

const DEFAULT_TYPING_INTERVAL_SECONDS: u32 = 3;
const MAX_TYPING_REFRESH_BACKOFF_SECONDS: u64 = 30;
const ACTIVITY_DISPATCH_BACKLOG_WARNING_THRESHOLD: usize = 64;
const ACTIVITY_BLOCKING_IO_MAX_SECS: u64 = 5;
const ACTIVITY_DISPATCH_SHUTDOWN_HEADROOM_MS: u64 = 500;
const READ_RECEIPT_RETRY_DELAY_MS: u64 = 5_000;
const READ_RECEIPT_OWNERSHIP_HIGH_WATERMARK: usize = 10_000;
const READ_RECEIPT_TASK_KIND: &str = "activityReadReceiptIngestV2";
// This budget must stay at or above the longest built-in activity operation
// timeout so graceful shutdown drains already-queued work instead of routinely
// dropping it. Built-in channel activity implementations must keep their own
// blocking I/O bounded within this shared ceiling.
const ACTIVITY_DISPATCH_SHUTDOWN_GRACE_MS: u64 =
    ACTIVITY_BLOCKING_IO_MAX_SECS * 1000 + ACTIVITY_DISPATCH_SHUTDOWN_HEADROOM_MS;
const _: () = {
    assert!(
        ACTIVITY_BLOCKING_IO_MAX_SECS >= crate::channels::signal::SIGNAL_HTTP_TYPING_TIMEOUT_SECS
    );
    assert!(
        ACTIVITY_BLOCKING_IO_MAX_SECS >= crate::channels::signal::SIGNAL_HTTP_RECEIPT_TIMEOUT_SECS
    );
    assert!(
        ACTIVITY_DISPATCH_SHUTDOWN_GRACE_MS
            > crate::channels::signal::SIGNAL_HTTP_TYPING_TIMEOUT_SECS * 1000
    );
    assert!(
        ACTIVITY_DISPATCH_SHUTDOWN_GRACE_MS
            > crate::channels::signal::SIGNAL_HTTP_RECEIPT_TIMEOUT_SECS * 1000
    );
};
const UNSUPPORTED_ACTIVITY_WARNING_COOLDOWN_SECS: u64 = 300;
const STOP_TYPING_THREAD_NAME: &str = "carapace-stop-typing";
// Stop-state machine:
// - NOT_REQUESTED -> TASK_RESERVED -> TASK_SPAWNING -> TASK_RUNNING -> COMPLETED
// - NOT_REQUESTED -> FALLBACK_RESERVED -> COMPLETED
// - TASK_RESERVED -> FALLBACK_RESERVED -> COMPLETED
// - TASK_SPAWNING -> FALLBACK_RESERVED -> COMPLETED
// TASK_SPAWNING means the typing task is detaching a stop worker, but fallback
// cleanup may still steal ownership if the outer task is aborted before that
// worker starts running. TASK_RUNNING means the detached stop worker has
// durably claimed exclusive ownership of the final stop_typing call.
const STOP_STATE_NOT_REQUESTED: u8 = 0;
const STOP_STATE_TASK_RESERVED: u8 = 1;
const STOP_STATE_TASK_SPAWNING: u8 = 2;
const STOP_STATE_TASK_RUNNING: u8 = 3;
const STOP_STATE_FALLBACK_RESERVED: u8 = 4;
const STOP_STATE_COMPLETED: u8 = 5;

#[derive(Debug, Error)]
pub enum ActivityStartupError {
    #[error("{source}")]
    ThreadSpawn {
        #[source]
        source: StartupThreadSpawnError,
    },
}

impl From<StartupThreadSpawnError> for ActivityStartupError {
    fn from(source: StartupThreadSpawnError) -> Self {
        Self::ThreadSpawn { source }
    }
}

struct UnsupportedActivityWarningRegistry {
    seen_at: HashMap<String, Instant>,
    cooldown: Duration,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct ReadReceiptTaskPayload {
    kind: String,
    channel_id: String,
    context: ReadReceiptContext,
}

impl ReadReceiptTaskPayload {
    fn new(channel_id: &str, context: ReadReceiptContext) -> Self {
        Self {
            kind: READ_RECEIPT_TASK_KIND.to_string(),
            channel_id: channel_id.to_string(),
            context,
        }
    }

    fn from_value(value: serde_json::Value) -> Result<Self, String> {
        let payload: Self = serde_json::from_value(value).map_err(|err| err.to_string())?;
        if payload.kind != READ_RECEIPT_TASK_KIND {
            return Err(format!(
                "unsupported read receipt task kind '{}'; expected '{}'. If this came from a persisted pre-upgrade task, remove or repair activity/read_receipts.json",
                payload.kind, READ_RECEIPT_TASK_KIND
            ));
        }
        Ok(payload)
    }
}

impl Default for UnsupportedActivityWarningRegistry {
    fn default() -> Self {
        Self {
            seen_at: HashMap::new(),
            cooldown: Duration::from_secs(UNSUPPORTED_ACTIVITY_WARNING_COOLDOWN_SECS),
        }
    }
}

impl UnsupportedActivityWarningRegistry {
    fn should_warn(&mut self, key: &str, now: Instant) -> bool {
        self.seen_at
            .retain(|_, last_seen| now.saturating_duration_since(*last_seen) < self.cooldown);
        match self.seen_at.get(key) {
            Some(_) => false,
            _ => {
                self.seen_at.insert(key.to_string(), now);
                true
            }
        }
    }

    #[cfg(test)]
    fn with_cooldown_for_test(cooldown: Duration) -> Self {
        Self {
            seen_at: HashMap::new(),
            cooldown,
        }
    }

    #[cfg(test)]
    fn reset(&mut self) {
        self.seen_at.clear();
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "kebab-case")]
pub enum TypingIndicatorMode {
    #[default]
    Thinking,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TypingFeaturePolicy {
    pub enabled: bool,
    pub mode: TypingIndicatorMode,
    pub interval_seconds: u32,
}

impl Default for TypingFeaturePolicy {
    fn default() -> Self {
        Self {
            enabled: false,
            mode: TypingIndicatorMode::Thinking,
            interval_seconds: DEFAULT_TYPING_INTERVAL_SECONDS,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "camelCase")]
pub struct ReadReceiptFeaturePolicy {
    pub enabled: bool,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "camelCase")]
pub struct ChannelActivityPolicy {
    pub typing: TypingFeaturePolicy,
    pub read_receipts: ReadReceiptFeaturePolicy,
}

/// Receive-time explicit read-receipt ownership claim.
///
/// Signal receive decides whether Carapace owns the receipt for this inbound
/// message. The claim is resolved at inbound append time and, when the
/// activity service issued it, also reserves bounded receipt ownership
/// capacity until it is released or completed.
#[derive(Debug, Clone)]
pub struct ClaimedReadReceipt {
    channel_id: String,
    context: ReadReceiptContext,
    reservation_id: Option<String>,
}

impl ClaimedReadReceipt {
    pub fn new(channel_id: impl Into<String>, context: ReadReceiptContext) -> Self {
        Self {
            channel_id: channel_id.into(),
            context,
            reservation_id: None,
        }
    }

    fn tracked(
        channel_id: impl Into<String>,
        context: ReadReceiptContext,
        reservation_id: impl Into<String>,
    ) -> Self {
        Self {
            channel_id: channel_id.into(),
            context,
            reservation_id: Some(reservation_id.into()),
        }
    }

    pub fn channel_id(&self) -> &str {
        &self.channel_id
    }

    pub fn context(&self) -> &ReadReceiptContext {
        &self.context
    }

    fn reservation_id(&self) -> Option<&str> {
        self.reservation_id.as_deref()
    }
}

/// Poll-scoped reservation of bounded read-receipt ownership capacity.
///
/// Signal receive reserves ownership slots before it suppresses upstream
/// auto-read-receipts. Each inbound message that Carapace chooses to own
/// consumes one reserved slot into a `ClaimedReadReceipt`, and any unused slots
/// are released automatically when the reservation is dropped at the end of the
/// poll.
pub struct ReadReceiptOwnershipReservation {
    channel_id: String,
    reservation_ids: Vec<String>,
    reserved_read_receipt_ownership: Arc<Mutex<HashSet<String>>>,
}

impl ReadReceiptOwnershipReservation {
    fn new(
        channel_id: impl Into<String>,
        reservation_ids: Vec<String>,
        reserved_read_receipt_ownership: Arc<Mutex<HashSet<String>>>,
    ) -> Self {
        Self {
            channel_id: channel_id.into(),
            reservation_ids,
            reserved_read_receipt_ownership,
        }
    }

    pub fn reserved_capacity(&self) -> usize {
        self.reservation_ids.len()
    }

    pub fn claim(&mut self, context: ReadReceiptContext) -> Option<ClaimedReadReceipt> {
        let reservation_id = self.reservation_ids.pop()?;
        Some(ClaimedReadReceipt::tracked(
            self.channel_id.clone(),
            context,
            reservation_id,
        ))
    }
}

impl Drop for ReadReceiptOwnershipReservation {
    fn drop(&mut self) {
        if self.reservation_ids.is_empty() {
            return;
        }

        let mut reserved_read_receipt_ownership = self.reserved_read_receipt_ownership.lock();
        for reservation_id in self.reservation_ids.drain(..) {
            reserved_read_receipt_ownership.remove(&reservation_id);
        }
    }
}

pub struct TypingLoopHandle {
    cancel: CancellationToken,
    task: Option<JoinHandle<()>>,
    runtime_handle: Handle,
    plugin: Arc<dyn ChannelPluginInstance>,
    ctx: TypingContext,
    channel_id: String,
    stop_state: Arc<AtomicU8>,
    activity_dispatcher: Arc<ActivityDispatcher>,
}

struct StopTypingDispatchRequest {
    plugin: Arc<dyn ChannelPluginInstance>,
    channel_id: String,
    ctx: TypingContext,
    stop_states: Vec<Arc<AtomicU8>>,
    queued: bool,
    in_flight: bool,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct StopTypingDispatchKey {
    channel_id: String,
    recipient: String,
    thread_id: Option<String>,
    account_id: Option<String>,
}

impl StopTypingDispatchKey {
    fn new(channel_id: &str, ctx: &TypingContext) -> Self {
        Self {
            channel_id: channel_id.to_string(),
            recipient: ctx.to.clone(),
            thread_id: ctx.thread_id.clone(),
            account_id: ctx.account_id.clone(),
        }
    }
}

pub struct ActivityService {
    dispatcher: Arc<ActivityDispatcher>,
    read_receipt_queue: Arc<TaskQueue>,
    read_receipt_wake: Arc<tokio::sync::Notify>,
    reserved_read_receipt_ownership: Arc<Mutex<HashSet<String>>>,
    unsupported_feature_warnings: Mutex<UnsupportedActivityWarningRegistry>,
    read_receipt_ownership_high_watermark: usize,
}

impl Default for ActivityService {
    fn default() -> Self {
        Self::new()
    }
}

impl ActivityService {
    pub fn new() -> Self {
        Self::try_new().expect("failed to initialize in-memory activity service startup helpers")
    }

    pub fn try_new() -> Result<Self, ActivityStartupError> {
        Self::try_with_read_receipt_queue(Arc::new(TaskQueue::with_capacity_limit(
            None,
            Some(READ_RECEIPT_OWNERSHIP_HIGH_WATERMARK),
        )))
    }

    pub fn new_persistent(state_dir: PathBuf) -> Self {
        Self::try_new_persistent(state_dir)
            .expect("failed to initialize persistent activity service startup helpers")
    }

    pub fn try_new_persistent(state_dir: PathBuf) -> Result<Self, ActivityStartupError> {
        Self::try_with_read_receipt_queue(Arc::new(TaskQueue::with_capacity_limit(
            Some(state_dir.join("activity").join("read_receipts.json")),
            Some(READ_RECEIPT_OWNERSHIP_HIGH_WATERMARK),
        )))
    }

    fn try_with_read_receipt_queue(
        read_receipt_queue: Arc<TaskQueue>,
    ) -> Result<Self, ActivityStartupError> {
        Ok(Self {
            dispatcher: Arc::new(ActivityDispatcher::try_new()?),
            read_receipt_queue,
            read_receipt_wake: Arc::new(tokio::sync::Notify::new()),
            reserved_read_receipt_ownership: Arc::new(Mutex::new(HashSet::new())),
            unsupported_feature_warnings: Mutex::new(UnsupportedActivityWarningRegistry::default()),
            read_receipt_ownership_high_watermark: READ_RECEIPT_OWNERSHIP_HIGH_WATERMARK,
        })
    }

    #[cfg(test)]
    fn with_test_components(
        read_receipt_queue: Arc<TaskQueue>,
        backlog_warning_threshold: usize,
        read_receipt_ownership_high_watermark: usize,
    ) -> Self {
        Self {
            dispatcher: Arc::new(ActivityDispatcher::with_backlog_warning_threshold(
                backlog_warning_threshold,
            )),
            read_receipt_queue,
            read_receipt_wake: Arc::new(tokio::sync::Notify::new()),
            reserved_read_receipt_ownership: Arc::new(Mutex::new(HashSet::new())),
            unsupported_feature_warnings: Mutex::new(UnsupportedActivityWarningRegistry::default()),
            read_receipt_ownership_high_watermark,
        }
    }

    #[cfg(test)]
    pub(crate) fn with_read_receipt_queue_for_test(read_receipt_queue: Arc<TaskQueue>) -> Self {
        Self::with_test_components(
            read_receipt_queue,
            ACTIVITY_DISPATCH_BACKLOG_WARNING_THRESHOLD,
            READ_RECEIPT_OWNERSHIP_HIGH_WATERMARK,
        )
    }

    #[cfg(test)]
    pub(crate) fn with_read_receipt_queue_and_limits_for_test(
        read_receipt_queue: Arc<TaskQueue>,
        backlog_warning_threshold: usize,
        read_receipt_ownership_high_watermark: usize,
    ) -> Self {
        Self::with_test_components(
            read_receipt_queue,
            backlog_warning_threshold,
            read_receipt_ownership_high_watermark,
        )
    }

    #[cfg(test)]
    pub(crate) fn with_limits_for_test(
        backlog_warning_threshold: usize,
        read_receipt_ownership_high_watermark: usize,
    ) -> Self {
        Self::with_test_components(
            Arc::new(TaskQueue::with_capacity_limit(
                None,
                Some(read_receipt_ownership_high_watermark),
            )),
            backlog_warning_threshold,
            read_receipt_ownership_high_watermark,
        )
    }

    pub fn dispatcher(&self) -> &Arc<ActivityDispatcher> {
        &self.dispatcher
    }

    pub fn read_receipt_queue(&self) -> &Arc<TaskQueue> {
        &self.read_receipt_queue
    }

    pub fn try_claim_read_receipt(
        &self,
        channel_id: &str,
        ctx: ReadReceiptContext,
    ) -> Option<ClaimedReadReceipt> {
        let mut reservation = self.reserve_read_receipt_ownership_slots(channel_id, 1)?;
        reservation.claim(ctx)
    }

    pub fn reserve_available_read_receipt_ownership(
        &self,
        channel_id: &str,
    ) -> Option<ReadReceiptOwnershipReservation> {
        self.reserve_read_receipt_ownership_slots(
            channel_id,
            self.read_receipt_ownership_high_watermark,
        )
    }

    fn reserve_read_receipt_ownership_slots(
        &self,
        channel_id: &str,
        requested_slots: usize,
    ) -> Option<ReadReceiptOwnershipReservation> {
        if requested_slots == 0 {
            return None;
        }

        let outstanding_queue = self.read_receipt_queue_outstanding_count();
        let mut reserved_read_receipt_ownership = self.reserved_read_receipt_ownership.lock();
        let outstanding = outstanding_queue + reserved_read_receipt_ownership.len();
        if outstanding >= self.read_receipt_ownership_high_watermark {
            drop(reserved_read_receipt_ownership);
            self.maybe_warn_read_receipt_backpressure(channel_id, outstanding);
            return None;
        }

        let available_slots = self.read_receipt_ownership_high_watermark - outstanding;
        let reserved_slots = available_slots.min(requested_slots);
        let reservation_ids = (0..reserved_slots)
            .map(|_| {
                let reservation_id = Uuid::new_v4().to_string();
                reserved_read_receipt_ownership.insert(reservation_id.clone());
                reservation_id
            })
            .collect();
        Some(ReadReceiptOwnershipReservation::new(
            channel_id,
            reservation_ids,
            self.reserved_read_receipt_ownership.clone(),
        ))
    }

    fn release_claimed_read_receipt_reservation(&self, claim: &ClaimedReadReceipt) -> bool {
        let Some(reservation_id) = claim.reservation_id() else {
            return false;
        };
        self.reserved_read_receipt_ownership
            .lock()
            .remove(reservation_id)
    }

    pub fn withhold_claimed_read_receipt(&self, claim: &ClaimedReadReceipt) -> bool {
        self.release_claimed_read_receipt_reservation(claim)
    }

    pub async fn complete_claimed_read_receipt(
        &self,
        state: &WsServerState,
        claim: &ClaimedReadReceipt,
    ) -> Result<(), String> {
        let _ = self.release_claimed_read_receipt_reservation(claim);
        self.acknowledge_read_receipt_now(state, claim.channel_id(), claim.context().clone())
            .await
    }

    async fn persist_ready_read_receipt_task(
        &self,
        channel_id: &str,
        ctx: ReadReceiptContext,
    ) -> crate::tasks::DurableTask {
        self.read_receipt_queue
            .enqueue_async(
                serde_json::to_value(ReadReceiptTaskPayload::new(channel_id, ctx))
                    .expect("read receipt task payload should serialize"),
                None,
            )
            .await
    }

    async fn persist_running_read_receipt_task(
        &self,
        channel_id: &str,
        ctx: ReadReceiptContext,
    ) -> crate::tasks::DurableTask {
        self.read_receipt_queue
            .enqueue_running_async_with_policy(
                serde_json::to_value(ReadReceiptTaskPayload::new(channel_id, ctx))
                    .expect("read receipt task payload should serialize"),
                crate::tasks::TaskPolicy::default(),
            )
            .await
    }

    pub async fn enqueue_ready_read_receipt(
        &self,
        channel_id: &str,
        ctx: ReadReceiptContext,
    ) -> Option<String> {
        let task = self.persist_ready_read_receipt_task(channel_id, ctx).await;
        if task.state == crate::tasks::TaskState::Failed {
            tracing::warn!(
                channel = %channel_id,
                error = ?task.last_error,
                "failed to persist immediate read receipt obligation"
            );
            None
        } else {
            // Any persisted non-failed enqueue may have made work runnable.
            // Spurious wakes are harmless and simpler than coupling this path
            // to specific task states.
            self.read_receipt_wake.notify_one();
            Some(task.id)
        }
    }

    async fn finalize_read_receipt_task(
        &self,
        task_id: &str,
        outcome: &TaskExecutionOutcome,
    ) -> Result<(), String> {
        let updated = match outcome {
            TaskExecutionOutcome::Done { run_id } => {
                tokio::task::spawn_blocking({
                    let queue = self.read_receipt_queue.clone();
                    let task_id = task_id.to_string();
                    let run_id = run_id.clone();
                    move || queue.mark_done(&task_id, run_id.as_deref())
                })
                .await
            }
            TaskExecutionOutcome::RetryWait { delay_ms, error } => {
                tokio::task::spawn_blocking({
                    let queue = self.read_receipt_queue.clone();
                    let task_id = task_id.to_string();
                    let error = error.clone();
                    let delay_ms = *delay_ms;
                    move || queue.mark_retry_wait(&task_id, delay_ms, &error)
                })
                .await
            }
            TaskExecutionOutcome::Blocked { reason, category } => {
                tokio::task::spawn_blocking({
                    let queue = self.read_receipt_queue.clone();
                    let task_id = task_id.to_string();
                    let reason = reason.clone();
                    let category = *category;
                    move || queue.mark_blocked(&task_id, &reason, category)
                })
                .await
            }
            TaskExecutionOutcome::Failed { error } => {
                tokio::task::spawn_blocking({
                    let queue = self.read_receipt_queue.clone();
                    let task_id = task_id.to_string();
                    let error = error.clone();
                    move || queue.mark_failed(&task_id, &error)
                })
                .await
            }
            TaskExecutionOutcome::Cancelled { reason } => {
                tokio::task::spawn_blocking({
                    let queue = self.read_receipt_queue.clone();
                    let task_id = task_id.to_string();
                    let reason = reason.clone();
                    move || queue.mark_cancelled(&task_id, reason.as_deref())
                })
                .await
            }
        }
        .map_err(|err| format!("read receipt inline-state worker failed: {err}"))?;

        if updated {
            Ok(())
        } else {
            Err(format!(
                "failed to update read receipt task '{}' after inline dispatch",
                task_id
            ))
        }
    }

    fn inline_read_receipt_result(
        outcome: TaskExecutionOutcome,
        finalize_result: Result<(), String>,
    ) -> Result<(), String> {
        match (outcome, finalize_result) {
            (TaskExecutionOutcome::Done { .. }, Ok(())) => Ok(()),
            (TaskExecutionOutcome::Done { .. }, Err(err)) => Err(format!(
                "read receipt was sent but durable task finalization failed: {err}"
            )),
            (TaskExecutionOutcome::RetryWait { error, .. }, Ok(())) => Err(error),
            (TaskExecutionOutcome::RetryWait { error, .. }, Err(err)) => Err(format!(
                "{error}; additionally failed to persist the retry schedule: {err}"
            )),
            (TaskExecutionOutcome::Blocked { reason, .. }, Ok(())) => Err(reason),
            (TaskExecutionOutcome::Blocked { reason, .. }, Err(err)) => Err(format!(
                "{reason}; additionally failed to persist the blocked state: {err}"
            )),
            (TaskExecutionOutcome::Failed { error }, Ok(())) => Err(error),
            (TaskExecutionOutcome::Failed { error }, Err(err)) => Err(format!(
                "{error}; additionally failed to persist the failure state: {err}"
            )),
            (TaskExecutionOutcome::Cancelled { reason }, Ok(())) => {
                Err(reason.unwrap_or_else(|| {
                    "read receipt task was cancelled during immediate dispatch".to_string()
                }))
            }
            (TaskExecutionOutcome::Cancelled { reason }, Err(err)) => Err(format!(
                "{}; additionally failed to persist the cancellation state: {err}",
                reason.unwrap_or_else(|| {
                    "read receipt task was cancelled during immediate dispatch".to_string()
                })
            )),
        }
    }

    pub async fn acknowledge_read_receipt_now(
        &self,
        state: &WsServerState,
        channel_id: &str,
        ctx: ReadReceiptContext,
    ) -> Result<(), String> {
        let task = self
            .persist_running_read_receipt_task(channel_id, ctx.clone())
            .await;
        if task.state != crate::tasks::TaskState::Failed {
            let task_id = task.id.clone();
            let outcome = execute_read_receipt_task(state, task).await;
            let finalize_result = self.finalize_read_receipt_task(&task_id, &outcome).await;
            return Self::inline_read_receipt_result(outcome, finalize_result);
        }

        tracing::warn!(
            channel = %channel_id,
            "failed to persist an immediate read receipt obligation; attempting direct explicit receipt send"
        );
        send_read_receipt_immediately(state, channel_id, ctx).await
    }

    pub fn dispatch_stop_typing(
        &self,
        plugin: Arc<dyn ChannelPluginInstance>,
        channel_id: &str,
        ctx: TypingContext,
        stop_state: Arc<AtomicU8>,
    ) {
        self.dispatcher
            .dispatch_stop_typing(plugin, channel_id, ctx, stop_state);
    }

    pub fn warn_unsupported_feature(&self, channel_id: &str, feature: &str) {
        let key = format!("{channel_id}:{feature}");
        let should_warn = {
            let mut registry = self.unsupported_feature_warnings.lock();
            registry.should_warn(&key, Instant::now())
        };
        if should_warn {
            tracing::warn!(
                channel = %channel_id,
                feature,
                "channel activity feature is enabled in config but unsupported by this channel; ignoring"
            );
        }
    }

    fn read_receipt_queue_outstanding_count(&self) -> usize {
        let stats = self.read_receipt_queue.stats();
        stats.queued + stats.running + stats.blocked + stats.retry_wait
    }

    fn read_receipt_obligation_count(&self) -> usize {
        self.read_receipt_queue_outstanding_count()
            + self.reserved_read_receipt_ownership.lock().len()
    }

    fn maybe_warn_read_receipt_backpressure(&self, channel_id: &str, outstanding: usize) {
        let key = format!("{channel_id}:read_receipts_backpressure");
        let should_warn = {
            let mut registry = self.unsupported_feature_warnings.lock();
            registry.should_warn(&key, Instant::now())
        };
        if should_warn {
            tracing::warn!(
                channel = %channel_id,
                outstanding,
                high_watermark = self.read_receipt_ownership_high_watermark,
                "read receipt backlog reached the high-water mark; leaving upstream auto-receipts enabled for new messages until the durable queue drains"
            );
        }
    }

    pub fn available_read_receipt_ownership_capacity(&self, channel_id: &str) -> usize {
        let outstanding = self.read_receipt_obligation_count();
        if outstanding < self.read_receipt_ownership_high_watermark {
            return self.read_receipt_ownership_high_watermark - outstanding;
        }

        self.maybe_warn_read_receipt_backpressure(channel_id, outstanding);
        0
    }

    pub fn can_accept_read_receipt_ownership(&self, channel_id: &str) -> bool {
        self.available_read_receipt_ownership_capacity(channel_id) > 0
    }

    pub async fn shutdown(&self) {
        let dispatcher = self.dispatcher.clone();
        if let Err(err) = tokio::task::spawn_blocking(move || dispatcher.shutdown()).await {
            tracing::warn!(error = %err, "activity service shutdown worker failed");
        }
    }

    pub async fn warn_configured_unsupported_features_for_registered_channels(
        &self,
        plugin_registry: Arc<PluginRegistry>,
    ) {
        for (channel_id, feature) in
            collect_configured_unsupported_features_for_registered_channels(plugin_registry).await
        {
            self.warn_unsupported_feature(&channel_id, feature);
        }
    }

    #[cfg(test)]
    pub(crate) fn reset_unsupported_activity_feature_warnings_for_test(&self) {
        self.unsupported_feature_warnings.lock().reset();
    }

    pub fn spawn_read_receipt_worker(
        &self,
        state: Arc<WsServerState>,
        shutdown: tokio::sync::watch::Receiver<bool>,
    ) {
        let queue = self.read_receipt_queue.clone();
        let executor = Arc::new(ReadReceiptTaskExecutor { state });
        let wake = self.read_receipt_wake.clone();
        tokio::spawn(crate::tasks::task_worker_loop_with_wakeup(
            queue,
            executor,
            Duration::from_secs(1),
            shutdown,
            Some(wake),
        ));
    }
}

pub struct ActivityDispatcher {
    stop_typing_tx: Mutex<Option<sync_mpsc::Sender<StopTypingDispatchKey>>>,
    stop_typing_worker: Mutex<Option<thread::JoinHandle<()>>>,
    stop_typing_pending: Arc<Mutex<HashMap<StopTypingDispatchKey, StopTypingDispatchRequest>>>,
    stop_typing_backlog: Arc<AtomicUsize>,
    backlog_warning_threshold: usize,
    shutting_down: Arc<AtomicBool>,
    shutdown_deadline: Arc<Mutex<Option<Instant>>>,
}

impl Default for ActivityDispatcher {
    fn default() -> Self {
        Self::new()
    }
}

impl ActivityDispatcher {
    pub fn new() -> Self {
        Self::try_new().expect("failed to initialize activity dispatcher startup thread")
    }

    pub fn try_new() -> Result<Self, ActivityStartupError> {
        Self::try_with_options(ACTIVITY_DISPATCH_BACKLOG_WARNING_THRESHOLD)
    }

    #[cfg(test)]
    pub(crate) fn with_backlog_warning_threshold(backlog_warning_threshold: usize) -> Self {
        Self::try_with_options(backlog_warning_threshold)
            .expect("failed to initialize activity dispatcher test worker")
    }

    #[cfg(test)]
    pub(crate) fn try_with_options_and_spawner_for_test(
        backlog_warning_threshold: usize,
        spawner: NamedThreadSpawner,
    ) -> Result<Self, ActivityStartupError> {
        Self::try_with_options_and_spawner(backlog_warning_threshold, spawner)
    }

    fn try_with_options(backlog_warning_threshold: usize) -> Result<Self, ActivityStartupError> {
        let spawn_default: NamedThreadSpawner = crate::thread_util::spawn_named_thread;
        Self::try_with_options_and_spawner(backlog_warning_threshold, spawn_default)
    }

    fn try_with_options_and_spawner(
        backlog_warning_threshold: usize,
        spawner: NamedThreadSpawner,
    ) -> Result<Self, ActivityStartupError> {
        let stop_typing_backlog = Arc::new(AtomicUsize::new(0));
        let shutting_down = Arc::new(AtomicBool::new(false));
        let shutdown_deadline = Arc::new(Mutex::new(None));

        let stop_typing_pending = Arc::new(Mutex::new(HashMap::<
            StopTypingDispatchKey,
            StopTypingDispatchRequest,
        >::new()));
        let stop_typing_pending_worker = stop_typing_pending.clone();
        let (stop_typing_tx_raw, stop_typing_rx) = sync_mpsc::channel::<StopTypingDispatchKey>();
        let stop_typing_tx = Mutex::new(Some(stop_typing_tx_raw));
        let stop_typing_backlog_worker = stop_typing_backlog.clone();
        let stop_typing_shutdown = shutting_down.clone();
        let stop_typing_deadline = shutdown_deadline.clone();
        let stop_typing_worker = spawn_startup_named_thread_with_spawner(
            STOP_TYPING_THREAD_NAME,
            Box::new(move || {
                // Stop-typing is cleanup, not optional side work. Keep this
                // path non-lossy, but coalesce requests per channel/recipient
                // so a stalled stop call cannot grow an unbounded duplicate
                // queue for the same remote typing state.
                while let Ok(key) = stop_typing_rx.recv() {
                    stop_typing_backlog_worker.fetch_sub(1, Ordering::AcqRel);
                    let maybe_request = {
                        let mut pending = stop_typing_pending_worker.lock();
                        let Some(request) = pending.get_mut(&key) else {
                            continue;
                        };
                        if should_drop_activity_work(
                            stop_typing_shutdown.as_ref(),
                            &stop_typing_deadline,
                        ) {
                            match pending.remove(&key) {
                                Some(request) => Some(Err(request)),
                                None => {
                                    tracing::warn!(
                                        "stop typing dispatcher lost a pending request during shutdown"
                                    );
                                    None
                                }
                            }
                        } else {
                            request.queued = false;
                            request.in_flight = true;
                            Some(Ok((
                                request.plugin.clone(),
                                request.channel_id.clone(),
                                request.ctx.clone(),
                                std::mem::take(&mut request.stop_states),
                            )))
                        }
                    };

                    let Some(request) = maybe_request else {
                        continue;
                    };

                    match request {
                        Err(request) => {
                            mark_stop_states_completed(&request.stop_states);
                            log_dropped_stop_typing_after_shutdown(
                                &request.channel_id,
                                &request.ctx,
                            );
                        }
                        Ok((mut plugin, mut channel_id, mut ctx, mut stop_states)) => loop {
                            dispatch_stop_typing_blocking(
                                plugin,
                                &channel_id,
                                ctx.clone(),
                                &stop_states,
                            );

                            let next_batch = {
                                let mut pending = stop_typing_pending_worker.lock();
                                let Some(request) = pending.get_mut(&key) else {
                                    break;
                                };
                                if should_drop_activity_work(
                                    stop_typing_shutdown.as_ref(),
                                    &stop_typing_deadline,
                                ) {
                                    match pending.remove(&key) {
                                        Some(request) => Some(Err(request)),
                                        None => {
                                            tracing::warn!(
                                                "stop typing dispatcher lost a pending request during shutdown"
                                            );
                                            None
                                        }
                                    }
                                } else if request.stop_states.is_empty() {
                                    pending.remove(&key);
                                    None
                                } else {
                                    Some(Ok((
                                        request.plugin.clone(),
                                        request.channel_id.clone(),
                                        request.ctx.clone(),
                                        std::mem::take(&mut request.stop_states),
                                    )))
                                }
                            };

                            match next_batch {
                                None => break,
                                Some(Err(request)) => {
                                    mark_stop_states_completed(&request.stop_states);
                                    log_dropped_stop_typing_after_shutdown(
                                        &request.channel_id,
                                        &request.ctx,
                                    );
                                    break;
                                }
                                Some(Ok((
                                    next_plugin,
                                    next_channel_id,
                                    next_ctx,
                                    next_stop_states,
                                ))) => {
                                    plugin = next_plugin;
                                    channel_id = next_channel_id;
                                    ctx = next_ctx;
                                    stop_states = next_stop_states;
                                }
                            }
                        },
                    }
                }
            }),
            spawner,
        )?;

        Ok(Self {
            stop_typing_tx,
            stop_typing_worker: Mutex::new(Some(stop_typing_worker)),
            stop_typing_pending,
            stop_typing_backlog,
            backlog_warning_threshold,
            shutting_down,
            shutdown_deadline,
        })
    }

    pub fn dispatch_stop_typing(
        &self,
        plugin: Arc<dyn ChannelPluginInstance>,
        channel_id: &str,
        ctx: TypingContext,
        stop_state: Arc<AtomicU8>,
    ) {
        let key = StopTypingDispatchKey::new(channel_id, &ctx);
        let mut should_enqueue = false;
        {
            let mut pending = self.stop_typing_pending.lock();
            if let Some(request) = pending.get_mut(&key) {
                request.stop_states.push(stop_state.clone());
                if !request.queued && !request.in_flight {
                    request.queued = true;
                    should_enqueue = true;
                }
            } else {
                pending.insert(
                    key.clone(),
                    StopTypingDispatchRequest {
                        plugin,
                        channel_id: channel_id.to_string(),
                        ctx,
                        stop_states: vec![stop_state.clone()],
                        queued: true,
                        in_flight: false,
                    },
                );
                should_enqueue = true;
            }
        }

        if !should_enqueue {
            return;
        }

        let Some(sender) = self.stop_typing_tx.lock().as_ref().cloned() else {
            let dropped = self.stop_typing_pending.lock().remove(&key);
            if let Some(request) = dropped {
                mark_stop_states_completed(&request.stop_states);
            } else {
                mark_stop_completed(stop_state.as_ref());
            }
            tracing::warn!(
                channel = %channel_id,
                "stop typing dispatcher is shut down; dropping implicit stop request"
            );
            return;
        };

        let backlog = self.stop_typing_backlog.fetch_add(1, Ordering::AcqRel) + 1;
        log_activity_backlog_if_needed(
            "stop typing",
            channel_id,
            backlog,
            self.backlog_warning_threshold,
        );

        if sender.send(key.clone()).is_err() {
            self.stop_typing_backlog.fetch_sub(1, Ordering::AcqRel);
            let dropped = self.stop_typing_pending.lock().remove(&key);
            if let Some(request) = dropped {
                mark_stop_states_completed(&request.stop_states);
            } else {
                mark_stop_completed(stop_state.as_ref());
            }
            tracing::warn!(
                channel = %channel_id,
                "stop typing dispatcher is shut down; dropping implicit stop request"
            );
        }
    }

    pub(crate) fn shutdown(&self) {
        self.shutdown_with_deadline(Duration::from_millis(ACTIVITY_DISPATCH_SHUTDOWN_GRACE_MS));
    }

    fn shutdown_with_deadline(&self, grace: Duration) {
        if self
            .shutting_down
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return;
        }
        let deadline = Instant::now() + grace;
        *self.shutdown_deadline.lock() = Some(deadline);
        self.stop_typing_tx.lock().take();

        if let Some(worker) = self.stop_typing_worker.lock().take() {
            join_activity_worker(worker, "stop typing");
        }
    }
}

impl Drop for ActivityDispatcher {
    fn drop(&mut self) {
        self.shutdown_with_deadline(Duration::ZERO);
    }
}

struct ReadReceiptTaskExecutor {
    state: Arc<WsServerState>,
}

fn read_receipt_policy_violation_error(task: &crate::tasks::DurableTask) -> Option<String> {
    if task.policy.max_attempts == 0 {
        Some("objective policy violation: maxAttempts must be greater than 0".to_string())
    } else if task.attempts > task.policy.max_attempts {
        Some(format!(
            "objective policy violation: attempts {} exceeded maxAttempts {}",
            task.attempts, task.policy.max_attempts
        ))
    } else {
        None
    }
}

async fn prepare_read_receipt_dispatch_plugin(
    state: &WsServerState,
    channel_id: &str,
) -> Result<Arc<dyn ChannelPluginInstance>, ReadReceiptDispatchError> {
    let Some(plugin_registry) = state.plugin_registry().cloned() else {
        return Err(ReadReceiptDispatchError::Retryable(
            "plugin registry unavailable for read receipt dispatch".to_string(),
        ));
    };

    let Some(plugin) = plugin_registry.get_channel(channel_id) else {
        return Err(ReadReceiptDispatchError::Retryable(format!(
            "channel plugin '{}' unavailable for read receipt dispatch",
            channel_id
        )));
    };

    match get_capabilities(plugin.clone()).await {
        Ok(capabilities) if capabilities.read_receipts => Ok(plugin),
        Ok(_) => {
            state
                .activity_service()
                .warn_unsupported_feature(channel_id, "read_receipts");
            Err(ReadReceiptDispatchError::Permanent(format!(
                "channel '{}' does not support read receipts after activation",
                channel_id
            )))
        }
        Err(err) => Err(ReadReceiptDispatchError::Retryable(format!(
            "failed to load channel capabilities for read receipt dispatch: {err}"
        ))),
    }
}

pub(crate) async fn send_read_receipt_immediately(
    state: &WsServerState,
    channel_id: &str,
    ctx: ReadReceiptContext,
) -> Result<(), String> {
    let plugin = prepare_read_receipt_dispatch_plugin(state, channel_id)
        .await
        .map_err(|err| match err {
            ReadReceiptDispatchError::Retryable(err) | ReadReceiptDispatchError::Permanent(err) => {
                err
            }
        })?;

    send_verified_read_receipt_with_plugin(plugin, channel_id, ctx)
        .await
        .map_err(|err| match err {
            ReadReceiptDispatchError::Retryable(err) | ReadReceiptDispatchError::Permanent(err) => {
                err
            }
        })
}

async fn execute_read_receipt_task(
    state: &WsServerState,
    task: crate::tasks::DurableTask,
) -> TaskExecutionOutcome {
    if let Some(error) = read_receipt_policy_violation_error(&task) {
        return TaskExecutionOutcome::Failed { error };
    }

    let payload = match ReadReceiptTaskPayload::from_value(task.payload) {
        Ok(payload) => payload,
        Err(err) => {
            return TaskExecutionOutcome::Failed {
                error: format!("invalid read receipt task payload: {err}"),
            };
        }
    };

    let plugin = match prepare_read_receipt_dispatch_plugin(state, &payload.channel_id).await {
        Ok(plugin) => plugin,
        Err(ReadReceiptDispatchError::Retryable(err)) => {
            return TaskExecutionOutcome::RetryWait {
                delay_ms: READ_RECEIPT_RETRY_DELAY_MS,
                error: err,
            };
        }
        Err(ReadReceiptDispatchError::Permanent(err)) => {
            return TaskExecutionOutcome::Failed { error: err };
        }
    };

    match send_verified_read_receipt_with_plugin(plugin, &payload.channel_id, payload.context).await
    {
        Ok(()) => TaskExecutionOutcome::Done { run_id: None },
        Err(ReadReceiptDispatchError::Retryable(err)) => TaskExecutionOutcome::RetryWait {
            delay_ms: READ_RECEIPT_RETRY_DELAY_MS,
            error: err,
        },
        Err(ReadReceiptDispatchError::Permanent(err)) => {
            TaskExecutionOutcome::Failed { error: err }
        }
    }
}

#[async_trait::async_trait]
impl TaskExecutor for ReadReceiptTaskExecutor {
    async fn execute(&self, task: crate::tasks::DurableTask) -> TaskExecutionOutcome {
        execute_read_receipt_task(self.state.as_ref(), task).await
    }
}

impl std::fmt::Debug for TypingLoopHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TypingLoopHandle")
            .field("channel_id", &self.channel_id)
            .finish_non_exhaustive()
    }
}

impl TypingLoopHandle {
    pub async fn stop(mut self) {
        self.cancel.cancel();
        if let Some(task) = self.task.take() {
            if let Err(err) = finish_typing_task(
                task,
                self.plugin.clone(),
                self.ctx.clone(),
                self.stop_state.clone(),
            )
            .await
            {
                tracing::warn!(
                    channel = %self.channel_id,
                    error = %err,
                    "failed to finish typing cleanup during explicit stop"
                );
            }
        }
    }
}

impl Drop for TypingLoopHandle {
    fn drop(&mut self) {
        if self.task.is_none() {
            return;
        }
        self.cancel.cancel();
        match Handle::try_current() {
            Ok(handle) if handle.runtime_flavor() == RuntimeFlavor::MultiThread => {
                if let Some(task) = self.task.take() {
                    let plugin = self.plugin.clone();
                    let ctx = self.ctx.clone();
                    let channel_id = self.channel_id.clone();
                    let stop_state = self.stop_state.clone();
                    if let Err(err) =
                        run_sync_blocking_send(finish_typing_task(task, plugin, ctx, stop_state))
                    {
                        tracing::warn!(
                            channel = %channel_id,
                            error = %err,
                            "failed to finish typing cleanup after drop"
                        );
                    }
                }
            }
            _ => self.drop_without_multithread_runtime(),
        }
    }
}

impl TypingLoopHandle {
    fn drop_without_multithread_runtime(&mut self) {
        if let Some(task) = self.task.take() {
            task.abort();
            observe_finished_typing_task_after_abort(
                task,
                self.runtime_handle.clone(),
                &self.channel_id,
            );
            if stop_fallback_needed(self.stop_state.as_ref()) {
                self.activity_dispatcher.dispatch_stop_typing(
                    self.plugin.clone(),
                    &self.channel_id,
                    self.ctx.clone(),
                    self.stop_state.clone(),
                );
            }
        }
    }
}

pub fn resolve_channel_activity_policy(config: &Value, channel: &str) -> ChannelActivityPolicy {
    let mut policy = ChannelActivityPolicy::default();
    apply_channel_activity_overrides(
        config
            .get("channels")
            .and_then(|channels| channels.get("defaults"))
            .and_then(|defaults| defaults.get("features")),
        &mut policy,
    );
    apply_channel_activity_overrides(
        config
            .get("channels")
            .and_then(|channels| channels.get(channel))
            .and_then(|entry| entry.get("features")),
        &mut policy,
    );
    if policy.typing.interval_seconds == 0 {
        policy.typing.interval_seconds = DEFAULT_TYPING_INTERVAL_SECONDS;
    }
    policy
}

pub fn load_channel_activity_policy(channel: &str) -> ChannelActivityPolicy {
    let config = match crate::config::load_raw_config_shared() {
        Ok(config) => config,
        Err(err) => {
            tracing::warn!(
                channel = %channel,
                error = %err,
                "failed to load config for channel activity policy; using disabled defaults"
            );
            Arc::new(serde_json::json!({}))
        }
    };
    resolve_channel_activity_policy(config.as_ref(), channel)
}

pub async fn load_channel_activity_policy_async(channel: &str) -> ChannelActivityPolicy {
    if let Some(config) = crate::config::peek_fresh_raw_config_shared() {
        return resolve_channel_activity_policy(config.as_ref(), channel);
    }

    let channel = channel.to_string();
    let worker_channel = channel.clone();
    match tokio::task::spawn_blocking(move || load_channel_activity_policy(&worker_channel)).await {
        Ok(policy) => policy,
        Err(err) => {
            tracing::warn!(
                channel = %channel,
                error = %err,
                "failed to load channel activity policy on blocking worker; using disabled defaults"
            );
            ChannelActivityPolicy::default()
        }
    }
}

pub async fn collect_configured_unsupported_features_for_registered_channels(
    plugin_registry: Arc<PluginRegistry>,
) -> Vec<(String, &'static str)> {
    let mut unsupported = Vec::new();
    for (channel_id, plugin) in plugin_registry.get_channels() {
        let policy = load_channel_activity_policy_async(&channel_id).await;
        if !policy.typing.enabled && !policy.read_receipts.enabled {
            continue;
        }

        let capabilities = match get_capabilities(plugin).await {
            Ok(capabilities) => capabilities,
            Err(err) => {
                tracing::warn!(
                    channel = %channel_id,
                    error = %err,
                    "failed to load channel capabilities while checking configured activity feature support"
                );
                continue;
            }
        };

        if policy.typing.enabled && !capabilities.typing_indicators {
            unsupported.push((channel_id.clone(), "typing"));
        }
        if policy.read_receipts.enabled && !capabilities.read_receipts {
            unsupported.push((channel_id, "read_receipts"));
        }
    }
    unsupported
}

fn apply_channel_activity_overrides(features: Option<&Value>, policy: &mut ChannelActivityPolicy) {
    let Some(features) = features.and_then(|value| value.as_object()) else {
        return;
    };

    if let Some(typing_value) = features.get("typing") {
        if let Some(typing) = typing_value.as_object() {
            if let Some(enabled) = typing.get("enabled").and_then(|value| value.as_bool()) {
                policy.typing.enabled = enabled;
            }
            if let Some(mode) = typing.get("mode").and_then(|value| value.as_str()) {
                if mode.eq_ignore_ascii_case("thinking") {
                    policy.typing.mode = TypingIndicatorMode::Thinking;
                } else {
                    tracing::warn!(
                        mode = %mode,
                        "unknown channels.*.features.typing.mode value; ignoring"
                    );
                }
            }
            if let Some(interval) = parse_positive_u32_from_value(typing_value, "intervalSeconds") {
                policy.typing.interval_seconds = interval;
            }
        }
    }

    if let Some(receipts) = features
        .get("readReceipts")
        .and_then(|value| value.as_object())
    {
        if let Some(enabled) = receipts.get("enabled").and_then(|value| value.as_bool()) {
            policy.read_receipts.enabled = enabled;
        }
    }
}

fn parse_positive_u32_from_value(value: &Value, key: &str) -> Option<u32> {
    value
        .get(key)
        .and_then(|entry| entry.as_u64())
        .filter(|entry| *entry > 0)
        .map(|entry| entry.min(u32::MAX as u64) as u32)
}

pub async fn maybe_start_typing_loop(
    plugin_registry: Arc<PluginRegistry>,
    channel_id: &str,
    policy: &ChannelActivityPolicy,
    prefetched_capabilities: Option<crate::plugins::ChannelCapabilities>,
    activity_service: Arc<ActivityService>,
    ctx: TypingContext,
) -> Option<TypingLoopHandle> {
    if !policy.typing.enabled {
        return None;
    }

    let plugin = plugin_registry.get_channel(channel_id)?;
    let capabilities = match prefetched_capabilities {
        Some(capabilities) => capabilities,
        None => get_capabilities(plugin.clone()).await.ok()?,
    };
    if !capabilities.typing_indicators {
        activity_service.warn_unsupported_feature(channel_id, "typing");
        return None;
    }

    if let Err(err) = invoke_start_typing(plugin.clone(), ctx.clone()).await {
        tracing::warn!(channel = %channel_id, error = %err, "failed to start typing indicator");
        return None;
    }

    let cancel = CancellationToken::new();
    let stop_state = Arc::new(AtomicU8::new(STOP_STATE_NOT_REQUESTED));
    let interval_seconds = policy.typing.interval_seconds.max(1);
    let runtime_handle = Handle::current();
    let channel_id = channel_id.to_string();
    let handle_channel_id = channel_id.clone();
    let handle_plugin = plugin.clone();
    let handle_ctx = ctx.clone();
    let handle_stop_state = stop_state.clone();
    let task_cancel = cancel.clone();
    let task_stop_state = stop_state.clone();
    let task = tokio::spawn(async move {
        let base_refresh_delay = Duration::from_secs(interval_seconds as u64);
        let mut consecutive_refresh_failures = 0_u32;
        let mut next_refresh_at = tokio::time::Instant::now() + base_refresh_delay;
        loop {
            tokio::select! {
                _ = task_cancel.cancelled() => {
                    break;
                }
                _ = tokio::time::sleep_until(next_refresh_at) => {
                    if let Err(err) = invoke_start_typing(plugin.clone(), ctx.clone()).await {
                        consecutive_refresh_failures = consecutive_refresh_failures.saturating_add(1);
                        let retry_delay = typing_refresh_retry_delay(
                            base_refresh_delay,
                            consecutive_refresh_failures,
                        );
                        next_refresh_at = tokio::time::Instant::now() + retry_delay;
                        if should_log_typing_refresh_failure(consecutive_refresh_failures) {
                            tracing::warn!(
                                channel = %channel_id,
                                error = %err,
                                failures = consecutive_refresh_failures,
                                retry_in_ms = retry_delay.as_millis(),
                                "failed to refresh typing indicator"
                            );
                        }
                    } else {
                        consecutive_refresh_failures = 0;
                        next_refresh_at = next_typing_refresh_deadline(
                            next_refresh_at,
                            base_refresh_delay,
                            tokio::time::Instant::now(),
                        );
                    }
                }
            }
        }

        if reserve_task_stop(task_stop_state.as_ref()) {
            if let Err(err) =
                invoke_stop_typing_with_running_state(plugin, ctx, task_stop_state).await
            {
                tracing::warn!(channel = %channel_id, error = %err, "failed to stop typing indicator");
            }
        }
    });

    Some(TypingLoopHandle {
        cancel,
        task: Some(task),
        runtime_handle,
        plugin: handle_plugin,
        ctx: handle_ctx,
        channel_id: handle_channel_id,
        stop_state: handle_stop_state,
        activity_dispatcher: activity_service.dispatcher().clone(),
    })
}

fn reserve_task_stop(stop_state: &AtomicU8) -> bool {
    stop_state
        .compare_exchange(
            STOP_STATE_NOT_REQUESTED,
            STOP_STATE_TASK_RESERVED,
            Ordering::AcqRel,
            Ordering::Acquire,
        )
        .is_ok()
}

fn stop_fallback_needed(stop_state: &AtomicU8) -> bool {
    let mut current = stop_state.load(Ordering::Acquire);
    loop {
        match current {
            STOP_STATE_NOT_REQUESTED | STOP_STATE_TASK_RESERVED | STOP_STATE_TASK_SPAWNING => {
                match stop_state.compare_exchange(
                    current,
                    STOP_STATE_FALLBACK_RESERVED,
                    Ordering::AcqRel,
                    Ordering::Acquire,
                ) {
                    Ok(_) => return true,
                    Err(next) => current = next,
                }
            }
            STOP_STATE_TASK_RUNNING | STOP_STATE_FALLBACK_RESERVED | STOP_STATE_COMPLETED => {
                return false;
            }
            _ => return false,
        }
    }
}

fn reserve_task_stop_worker_handoff(stop_state: &AtomicU8) -> bool {
    stop_state
        .compare_exchange(
            STOP_STATE_TASK_RESERVED,
            STOP_STATE_TASK_SPAWNING,
            Ordering::AcqRel,
            Ordering::Acquire,
        )
        .is_ok()
}

fn claim_task_stop_worker_ownership(stop_state: &AtomicU8) -> bool {
    stop_state
        .compare_exchange(
            STOP_STATE_TASK_SPAWNING,
            STOP_STATE_TASK_RUNNING,
            Ordering::AcqRel,
            Ordering::Acquire,
        )
        .is_ok()
}

fn mark_stop_completed(stop_state: &AtomicU8) {
    stop_state.store(STOP_STATE_COMPLETED, Ordering::Release);
}

fn mark_stop_states_completed(stop_states: &[Arc<AtomicU8>]) {
    for stop_state in stop_states {
        mark_stop_completed(stop_state.as_ref());
    }
}

async fn finish_typing_task(
    task: JoinHandle<()>,
    plugin: Arc<dyn ChannelPluginInstance>,
    ctx: TypingContext,
    stop_state: Arc<AtomicU8>,
) -> Result<(), String> {
    match task.await {
        Ok(()) => Ok(()),
        Err(join_err) => {
            if stop_fallback_needed(stop_state.as_ref()) {
                invoke_stop_typing(plugin, ctx)
                    .await
                    .map_err(|err| format!(
                        "typing task ended unexpectedly ({join_err}); fallback stop_typing also failed: {err}"
                    ))?;
                mark_stop_completed(stop_state.as_ref());
                Err(format!(
                    "typing task ended unexpectedly ({join_err}); sent fallback stop_typing"
                ))
            } else {
                Err(format!(
                    "typing task ended unexpectedly ({join_err}) after typing cleanup already completed"
                ))
            }
        }
    }
}

fn typing_refresh_retry_delay(base_delay: Duration, consecutive_failures: u32) -> Duration {
    if consecutive_failures == 0 {
        return base_delay;
    }
    let base_secs = base_delay.as_secs().max(1);
    let exponent = consecutive_failures.saturating_sub(1).min(4);
    let multiplier = 1_u64 << exponent;
    Duration::from_secs(
        base_secs
            .saturating_mul(multiplier)
            .min(MAX_TYPING_REFRESH_BACKOFF_SECONDS),
    )
}

fn next_typing_refresh_deadline(
    previous_deadline: tokio::time::Instant,
    cadence: Duration,
    now: tokio::time::Instant,
) -> tokio::time::Instant {
    let mut next_deadline = previous_deadline + cadence;
    while next_deadline <= now {
        next_deadline += cadence;
    }
    next_deadline
}

fn should_log_typing_refresh_failure(consecutive_failures: u32) -> bool {
    consecutive_failures <= 3 || consecutive_failures.is_power_of_two()
}

fn log_activity_backlog_if_needed(
    activity: &str,
    channel_id: &str,
    backlog: usize,
    threshold: usize,
) {
    if backlog >= threshold && backlog.is_power_of_two() {
        tracing::warn!(
            channel = %channel_id,
            backlog,
            activity,
            "activity dispatcher backlog is growing"
        );
    }
}

fn should_drop_activity_work(
    shutting_down: &AtomicBool,
    shutdown_deadline: &Mutex<Option<Instant>>,
) -> bool {
    if !shutting_down.load(Ordering::Acquire) {
        return false;
    }
    shutdown_deadline
        .lock()
        .as_ref()
        .is_some_and(|deadline| Instant::now() > *deadline)
}

fn log_dropped_stop_typing_after_shutdown(channel_id: &str, ctx: &TypingContext) {
    let _ = ctx;
    tracing::warn!(
        channel = %channel_id,
        "dropped queued stop-typing cleanup after activity shutdown deadline"
    );
}

fn observe_finished_typing_task_after_abort(
    task: JoinHandle<()>,
    runtime_handle: Handle,
    channel_id: &str,
) {
    let channel_id = channel_id.to_string();
    runtime_handle.spawn(async move {
        if let Err(err) = task.await {
            if err.is_panic() {
                tracing::warn!(
                    channel = %channel_id,
                    error = %panic_payload_to_string(err.into_panic()),
                    "typing task panicked during implicit drop cleanup"
                );
            }
        }
    });
}

fn join_activity_worker(worker: thread::JoinHandle<()>, activity: &'static str) {
    if let Err(payload) = worker.join() {
        tracing::warn!(
            activity,
            error = %panic_payload_to_string(payload),
            "activity dispatcher worker panicked during shutdown"
        );
    }
}

async fn invoke_stop_typing_with_running_state(
    plugin: Arc<dyn ChannelPluginInstance>,
    ctx: TypingContext,
    stop_state: Arc<AtomicU8>,
) -> Result<(), BindingError> {
    let Some(stop_task) = spawn_stop_typing_worker(plugin, ctx, stop_state) else {
        return Ok(());
    };
    let result = stop_task
        .await
        .map_err(|err| BindingError::CallError(err.to_string()))
        .and_then(|result| result);
    result
}

fn spawn_stop_typing_worker(
    plugin: Arc<dyn ChannelPluginInstance>,
    ctx: TypingContext,
    stop_state: Arc<AtomicU8>,
) -> Option<tokio::task::JoinHandle<Result<(), BindingError>>> {
    if !reserve_task_stop_worker_handoff(stop_state.as_ref()) {
        return None;
    }

    Some(tokio::task::spawn_blocking(move || {
        if !claim_task_stop_worker_ownership(stop_state.as_ref()) {
            return Ok(());
        }
        let result = catch_unwind(AssertUnwindSafe(|| plugin.stop_typing(ctx)));
        mark_stop_completed(stop_state.as_ref());
        match result {
            Ok(result) => result,
            Err(payload) => resume_unwind(payload),
        }
    }))
}

#[derive(Debug, Clone, PartialEq, Eq)]
enum ReadReceiptDispatchError {
    Retryable(String),
    Permanent(String),
}

async fn send_verified_read_receipt_with_plugin(
    plugin: Arc<dyn ChannelPluginInstance>,
    channel_id: &str,
    ctx: ReadReceiptContext,
) -> Result<(), ReadReceiptDispatchError> {
    let channel_id = channel_id.to_string();
    let worker_channel_id = channel_id.clone();
    match tokio::task::spawn_blocking(move || {
        dispatch_read_receipt_blocking(plugin, &worker_channel_id, ctx)
    })
    .await
    {
        Ok(result) => result,
        Err(err) if err.is_panic() => Err(ReadReceiptDispatchError::Permanent(format!(
            "read receipt worker task panicked: {}",
            panic_payload_to_string(err.into_panic())
        ))),
        Err(err) => Err(ReadReceiptDispatchError::Retryable(format!(
            "read receipt worker task failed: {err}"
        ))),
    }
}

async fn get_capabilities(
    plugin: Arc<dyn ChannelPluginInstance>,
) -> Result<crate::plugins::ChannelCapabilities, BindingError> {
    tokio::task::spawn_blocking(move || plugin.get_capabilities())
        .await
        .map_err(|err| BindingError::CallError(err.to_string()))?
}

async fn invoke_start_typing(
    plugin: Arc<dyn ChannelPluginInstance>,
    ctx: TypingContext,
) -> Result<(), BindingError> {
    tokio::task::spawn_blocking(move || plugin.start_typing(ctx))
        .await
        .map_err(|err| BindingError::CallError(err.to_string()))?
}

async fn invoke_stop_typing(
    plugin: Arc<dyn ChannelPluginInstance>,
    ctx: TypingContext,
) -> Result<(), BindingError> {
    tokio::task::spawn_blocking(move || plugin.stop_typing(ctx))
        .await
        .map_err(|err| BindingError::CallError(err.to_string()))?
}

fn dispatch_read_receipt_blocking(
    plugin: Arc<dyn ChannelPluginInstance>,
    channel_id: &str,
    ctx: ReadReceiptContext,
) -> Result<(), ReadReceiptDispatchError> {
    match catch_unwind(AssertUnwindSafe(|| plugin.mark_read(ctx))) {
        Ok(Ok(())) => Ok(()),
        Ok(Err(err)) => Err(ReadReceiptDispatchError::Retryable(format!(
            "failed to send read receipt on channel '{channel_id}': {err}"
        ))),
        Err(payload) => Err(ReadReceiptDispatchError::Permanent(format!(
            "read receipt dispatcher panicked on channel '{channel_id}': {}",
            panic_payload_to_string(payload)
        ))),
    }
}

fn dispatch_stop_typing_blocking(
    plugin: Arc<dyn ChannelPluginInstance>,
    channel_id: &str,
    ctx: TypingContext,
    stop_states: &[Arc<AtomicU8>],
) {
    let result = catch_unwind(AssertUnwindSafe(|| plugin.stop_typing(ctx)));
    mark_stop_states_completed(stop_states);
    match result {
        Ok(Ok(())) => {}
        Ok(Err(err)) => {
            tracing::warn!(
                channel = %channel_id,
                error = %err,
                "failed to stop typing indicator after drop"
            );
        }
        Err(payload) => {
            tracing::warn!(
                channel = %channel_id,
                error = %panic_payload_to_string(payload),
                "stop typing dispatcher panicked"
            );
        }
    }
}

fn panic_payload_to_string(payload: Box<dyn Any + Send + 'static>) -> String {
    let payload = payload.as_ref();
    if let Some(message) = payload.downcast_ref::<&str>() {
        (*message).to_string()
    } else if let Some(message) = payload.downcast_ref::<String>() {
        message.clone()
    } else {
        "non-string panic payload".to_string()
    }
}

#[cfg(test)]
mod tests {
    use std::error::Error as _;
    use std::fs;
    use std::sync::atomic::{AtomicU32, Ordering};

    use super::*;
    use crate::plugins::{ChannelCapabilities, ChannelInfo, DeliveryResult, OutboundContext};
    use tokio::sync::Notify;

    struct MockChannel {
        caps: ChannelCapabilities,
        start_typing_count: AtomicU32,
        stop_typing_count: AtomicU32,
        mark_read_count: AtomicU32,
        mark_read_started_notify: Option<Arc<Notify>>,
        mark_read_notify: Option<Arc<Notify>>,
        stop_typing_started_notify: Option<Arc<Notify>>,
        stop_typing_notify: Option<Arc<Notify>>,
        mark_read_delay: Duration,
        stop_typing_delay: Duration,
        panic_get_capabilities: bool,
        panic_mark_read_count: AtomicU32,
    }

    impl MockChannel {
        fn new(caps: ChannelCapabilities) -> Self {
            Self::with_stop_typing_notify(caps, None)
        }

        fn with_stop_typing_notify(
            caps: ChannelCapabilities,
            stop_typing_notify: Option<Arc<Notify>>,
        ) -> Self {
            Self {
                caps,
                start_typing_count: AtomicU32::new(0),
                stop_typing_count: AtomicU32::new(0),
                mark_read_count: AtomicU32::new(0),
                mark_read_started_notify: None,
                mark_read_notify: None,
                stop_typing_started_notify: None,
                stop_typing_notify,
                mark_read_delay: Duration::ZERO,
                stop_typing_delay: Duration::ZERO,
                panic_get_capabilities: false,
                panic_mark_read_count: AtomicU32::new(0),
            }
        }

        fn with_stop_typing_delay(caps: ChannelCapabilities, stop_typing_delay: Duration) -> Self {
            Self {
                stop_typing_delay,
                ..Self::with_stop_typing_notify(caps, None)
            }
        }

        fn with_stop_typing_delay_and_notify(
            caps: ChannelCapabilities,
            stop_typing_delay: Duration,
            stop_typing_notify: Arc<Notify>,
        ) -> Self {
            Self {
                stop_typing_delay,
                stop_typing_notify: Some(stop_typing_notify),
                ..Self::with_stop_typing_notify(caps, None)
            }
        }

        fn with_stop_typing_delay_and_started_notify(
            caps: ChannelCapabilities,
            stop_typing_delay: Duration,
            stop_typing_started_notify: Arc<Notify>,
        ) -> Self {
            Self {
                stop_typing_delay,
                stop_typing_started_notify: Some(stop_typing_started_notify),
                ..Self::with_stop_typing_notify(caps, None)
            }
        }

        fn with_panicking_mark_read(caps: ChannelCapabilities, panic_count: u32) -> Self {
            Self {
                panic_mark_read_count: AtomicU32::new(panic_count),
                ..Self::with_stop_typing_notify(caps, None)
            }
        }

        fn with_mark_read_delay(caps: ChannelCapabilities, mark_read_delay: Duration) -> Self {
            Self {
                mark_read_delay,
                ..Self::with_stop_typing_notify(caps, None)
            }
        }
    }

    impl ChannelPluginInstance for MockChannel {
        fn get_info(&self) -> Result<ChannelInfo, BindingError> {
            Ok(ChannelInfo {
                id: "mock".to_string(),
                label: "Mock".to_string(),
                selection_label: "Mock".to_string(),
                docs_path: String::new(),
                blurb: String::new(),
                order: 0,
            })
        }

        fn get_capabilities(&self) -> Result<crate::plugins::ChannelCapabilities, BindingError> {
            assert!(!self.panic_get_capabilities, "mock get_capabilities panic");
            Ok(self.caps.clone())
        }

        fn send_text(&self, _ctx: OutboundContext) -> Result<DeliveryResult, BindingError> {
            unreachable!()
        }

        fn send_media(&self, _ctx: OutboundContext) -> Result<DeliveryResult, BindingError> {
            unreachable!()
        }

        fn start_typing(&self, _ctx: TypingContext) -> Result<(), BindingError> {
            self.start_typing_count.fetch_add(1, Ordering::Relaxed);
            Ok(())
        }

        fn stop_typing(&self, _ctx: TypingContext) -> Result<(), BindingError> {
            if let Some(notify) = &self.stop_typing_started_notify {
                notify.notify_one();
            }
            if !self.stop_typing_delay.is_zero() {
                std::thread::sleep(self.stop_typing_delay);
            }
            self.stop_typing_count.fetch_add(1, Ordering::Relaxed);
            if let Some(notify) = &self.stop_typing_notify {
                notify.notify_one();
            }
            Ok(())
        }

        fn mark_read(&self, _ctx: ReadReceiptContext) -> Result<(), BindingError> {
            if let Some(notify) = &self.mark_read_started_notify {
                notify.notify_one();
            }
            if self.panic_mark_read_count.load(Ordering::Relaxed) > 0 {
                self.panic_mark_read_count.fetch_sub(1, Ordering::Relaxed);
                panic!("mock mark_read panic");
            }
            if !self.mark_read_delay.is_zero() {
                std::thread::sleep(self.mark_read_delay);
            }
            self.mark_read_count.fetch_add(1, Ordering::Relaxed);
            if let Some(notify) = &self.mark_read_notify {
                notify.notify_one();
            }
            Ok(())
        }
    }

    #[test]
    fn test_resolve_channel_activity_policy_defaults_disabled() {
        let policy = resolve_channel_activity_policy(&serde_json::json!({}), "signal");
        assert!(!policy.typing.enabled);
        assert_eq!(
            policy.typing.interval_seconds,
            DEFAULT_TYPING_INTERVAL_SECONDS
        );
        assert!(!policy.read_receipts.enabled);
    }

    #[test]
    fn test_resolve_channel_activity_policy_channel_override_wins() {
        let policy = resolve_channel_activity_policy(
            &serde_json::json!({
                "channels": {
                    "defaults": {
                        "features": {
                            "typing": {
                                "enabled": true,
                                "intervalSeconds": 5
                            },
                            "readReceipts": {
                                "enabled": false
                            }
                        }
                    },
                    "signal": {
                        "features": {
                            "typing": {
                                "enabled": false,
                                "intervalSeconds": 11
                            },
                            "readReceipts": {
                                "enabled": true
                            }
                        }
                    }
                }
            }),
            "signal",
        );
        assert!(!policy.typing.enabled);
        assert_eq!(policy.typing.interval_seconds, 11);
        assert!(policy.read_receipts.enabled);
    }

    #[test]
    fn test_load_channel_activity_policy_falls_back_to_defaults_on_load_error() {
        let _config_guard = crate::test_support::config::ScopedConfigCache::new();
        crate::config::clear_cache();
        let mut env_guard = crate::test_support::env::ScopedEnv::new();
        env_guard
            .set(
                "CARAPACE_CONFIG_PATH",
                std::path::Path::new("/definitely/missing/carapace.json5").as_os_str(),
            )
            .set("CARAPACE_DISABLE_CONFIG_CACHE", "1");

        let policy = load_channel_activity_policy("signal");
        assert!(!policy.typing.enabled);
        assert_eq!(
            policy.typing.interval_seconds,
            DEFAULT_TYPING_INTERVAL_SECONDS
        );
        assert!(!policy.read_receipts.enabled);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_load_channel_activity_policy_async_uses_cached_snapshot_without_disk_reload() {
        let _config_guard = crate::test_support::config::ScopedConfigCache::new();
        crate::config::clear_cache();
        let mut env_guard = crate::test_support::env::ScopedEnv::new();
        env_guard.set(
            "CARAPACE_CONFIG_PATH",
            std::path::Path::new("/definitely/missing/carapace.json5").as_os_str(),
        );

        crate::config::update_cache(
            serde_json::json!({
                "channels": {
                    "signal": {
                        "features": {
                            "typing": {
                                "enabled": true,
                                "intervalSeconds": 7
                            }
                        }
                    }
                }
            }),
            serde_json::json!({}),
        );

        let policy = load_channel_activity_policy_async("signal").await;
        assert!(policy.typing.enabled);
        assert_eq!(policy.typing.interval_seconds, 7);

        crate::config::clear_cache();
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_load_channel_activity_policy_async_refreshes_stale_cache_from_disk() {
        let _config_guard = crate::test_support::config::ScopedConfigCache::new();
        crate::config::clear_cache();
        let temp = tempfile::tempdir().unwrap();
        let config_path = temp.path().join("carapace.json5");
        fs::write(
            &config_path,
            r#"{
                "channels": {
                    "signal": {
                        "features": {
                            "typing": {
                                "enabled": false
                            }
                        }
                    }
                }
            }"#,
        )
        .unwrap();

        let mut env_guard = crate::test_support::env::ScopedEnv::new();
        env_guard
            .set("CARAPACE_CONFIG_PATH", config_path.as_os_str())
            .unset("CARAPACE_DISABLE_CONFIG_CACHE")
            .set("CARAPACE_CONFIG_CACHE_MS", "1");

        crate::config::update_cache_for_test_with_age(
            serde_json::json!({
                "channels": {
                    "signal": {
                        "features": {
                            "typing": {
                                "enabled": true
                            }
                        }
                    }
                }
            }),
            serde_json::json!({}),
            Duration::from_secs(1),
        );

        let policy = load_channel_activity_policy_async("signal").await;
        assert!(!policy.typing.enabled);

        crate::config::clear_cache();
    }

    #[tokio::test]
    async fn test_maybe_start_typing_loop_drives_start_and_stop() {
        let plugin = Arc::new(MockChannel::new(ChannelCapabilities {
            typing_indicators: true,
            ..Default::default()
        }));
        let registry = Arc::new(PluginRegistry::new());
        registry.register_channel("signal".to_string(), plugin.clone());
        let policy = ChannelActivityPolicy {
            typing: TypingFeaturePolicy {
                enabled: true,
                interval_seconds: 1,
                ..Default::default()
            },
            ..Default::default()
        };

        let activity_service = Arc::new(ActivityService::new());
        let handle = maybe_start_typing_loop(
            registry,
            "signal",
            &policy,
            None,
            activity_service.clone(),
            TypingContext {
                to: "+15551234567".to_string(),
                ..Default::default()
            },
        )
        .await
        .expect("typing loop should start");

        tokio::time::sleep(Duration::from_millis(30)).await;
        handle.stop().await;

        assert!(plugin.start_typing_count.load(Ordering::Relaxed) >= 1);
        assert_eq!(plugin.stop_typing_count.load(Ordering::Relaxed), 1);
        activity_service.shutdown().await;
    }

    #[tokio::test]
    async fn test_maybe_start_typing_loop_drop_cancels_background_task() {
        let stop_typing_notify = Arc::new(Notify::new());
        let plugin = Arc::new(MockChannel::with_stop_typing_notify(
            ChannelCapabilities {
                typing_indicators: true,
                ..Default::default()
            },
            Some(stop_typing_notify.clone()),
        ));
        let registry = Arc::new(PluginRegistry::new());
        registry.register_channel("signal".to_string(), plugin.clone());
        let policy = ChannelActivityPolicy {
            typing: TypingFeaturePolicy {
                enabled: true,
                interval_seconds: 1,
                ..Default::default()
            },
            ..Default::default()
        };

        let activity_service = Arc::new(ActivityService::new());
        let handle = maybe_start_typing_loop(
            registry,
            "signal",
            &policy,
            None,
            activity_service.clone(),
            TypingContext {
                to: "+15551234567".to_string(),
                ..Default::default()
            },
        )
        .await
        .expect("typing loop should start");

        drop(handle);
        tokio::time::timeout(Duration::from_secs(1), stop_typing_notify.notified())
            .await
            .expect("drop should stop typing promptly");

        assert!(plugin.start_typing_count.load(Ordering::Relaxed) >= 1);
        assert_eq!(plugin.stop_typing_count.load(Ordering::Relaxed), 1);
        activity_service.shutdown().await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_maybe_start_typing_loop_drop_cancels_background_task_current_thread() {
        let stop_typing_notify = Arc::new(Notify::new());
        let plugin = Arc::new(MockChannel::with_stop_typing_notify(
            ChannelCapabilities {
                typing_indicators: true,
                ..Default::default()
            },
            Some(stop_typing_notify.clone()),
        ));
        let registry = Arc::new(PluginRegistry::new());
        registry.register_channel("signal".to_string(), plugin.clone());
        let policy = ChannelActivityPolicy {
            typing: TypingFeaturePolicy {
                enabled: true,
                interval_seconds: 1,
                ..Default::default()
            },
            ..Default::default()
        };

        let activity_service = Arc::new(ActivityService::new());
        let handle = maybe_start_typing_loop(
            registry,
            "signal",
            &policy,
            None,
            activity_service.clone(),
            TypingContext {
                to: "+15551234567".to_string(),
                ..Default::default()
            },
        )
        .await
        .expect("typing loop should start");

        drop(handle);
        tokio::time::timeout(Duration::from_secs(1), stop_typing_notify.notified())
            .await
            .expect("drop should stop typing promptly on current-thread runtimes");
        tokio::task::yield_now().await;

        assert!(plugin.start_typing_count.load(Ordering::Relaxed) >= 1);
        assert_eq!(plugin.stop_typing_count.load(Ordering::Relaxed), 1);
        activity_service.shutdown().await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_maybe_start_typing_loop_drop_does_not_block_current_thread_runtime() {
        let stop_typing_notify = Arc::new(Notify::new());
        let plugin = Arc::new(MockChannel::with_stop_typing_delay_and_notify(
            ChannelCapabilities {
                typing_indicators: true,
                ..Default::default()
            },
            Duration::from_millis(250),
            stop_typing_notify.clone(),
        ));
        let registry = Arc::new(PluginRegistry::new());
        registry.register_channel("signal".to_string(), plugin.clone());
        let policy = ChannelActivityPolicy {
            typing: TypingFeaturePolicy {
                enabled: true,
                interval_seconds: 1,
                ..Default::default()
            },
            ..Default::default()
        };

        let activity_service = Arc::new(ActivityService::new());
        let handle = maybe_start_typing_loop(
            registry,
            "signal",
            &policy,
            None,
            activity_service.clone(),
            TypingContext {
                to: "+15551234567".to_string(),
                ..Default::default()
            },
        )
        .await
        .expect("typing loop should start");

        drop(handle);
        assert_eq!(
            plugin.stop_typing_count.load(Ordering::Relaxed),
            0,
            "drop should return before slow stop_typing completes"
        );
        tokio::time::timeout(Duration::from_secs(1), stop_typing_notify.notified())
            .await
            .expect("drop should still schedule stop_typing asynchronously");
        assert_eq!(plugin.stop_typing_count.load(Ordering::Relaxed), 1);
        activity_service.shutdown().await;
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_maybe_start_typing_loop_drop_stops_when_cleanup_was_only_reserved() {
        let stop_typing_notify = Arc::new(Notify::new());
        let plugin = Arc::new(MockChannel::with_stop_typing_notify(
            ChannelCapabilities {
                typing_indicators: true,
                ..Default::default()
            },
            Some(stop_typing_notify.clone()),
        ));
        let registry = Arc::new(PluginRegistry::new());
        registry.register_channel("signal".to_string(), plugin.clone());
        let policy = ChannelActivityPolicy {
            typing: TypingFeaturePolicy {
                enabled: true,
                interval_seconds: 1,
                ..Default::default()
            },
            ..Default::default()
        };

        let activity_service = Arc::new(ActivityService::new());
        let handle = maybe_start_typing_loop(
            registry,
            "signal",
            &policy,
            None,
            activity_service.clone(),
            TypingContext {
                to: "+15551234567".to_string(),
                ..Default::default()
            },
        )
        .await
        .expect("typing loop should start");

        handle
            .stop_state
            .store(STOP_STATE_TASK_RESERVED, Ordering::Release);
        drop(handle);
        tokio::time::timeout(Duration::from_secs(1), stop_typing_notify.notified())
            .await
            .expect("drop should still stop typing when task cleanup was only reserved");
        tokio::task::yield_now().await;

        assert_eq!(plugin.stop_typing_count.load(Ordering::Relaxed), 1);
        activity_service.shutdown().await;
    }

    #[tokio::test]
    async fn test_finish_typing_task_recovers_from_panicked_task() {
        let stop_typing_notify = Arc::new(Notify::new());
        let plugin = Arc::new(MockChannel::with_stop_typing_notify(
            ChannelCapabilities {
                typing_indicators: true,
                ..Default::default()
            },
            Some(stop_typing_notify.clone()),
        ));
        let task = tokio::spawn(async move {
            panic!("mock typing task panic");
        });

        let result = finish_typing_task(
            task,
            plugin.clone(),
            TypingContext {
                to: "+15551234567".to_string(),
                ..Default::default()
            },
            Arc::new(AtomicU8::new(STOP_STATE_NOT_REQUESTED)),
        )
        .await;

        let err = result.expect_err("panicked task should report fallback cleanup");
        assert!(
            err.contains("sent fallback stop_typing"),
            "expected fallback stop_typing diagnostic, got: {err}"
        );
        tokio::time::timeout(Duration::from_secs(1), stop_typing_notify.notified())
            .await
            .expect("panicked task should still trigger fallback stop_typing");

        assert_eq!(plugin.stop_typing_count.load(Ordering::Relaxed), 1);
    }

    #[test]
    fn test_stop_fallback_needed_skips_when_task_stop_worker_is_running() {
        let stop_state = AtomicU8::new(STOP_STATE_TASK_RUNNING);
        assert!(!stop_fallback_needed(&stop_state));
        assert_eq!(stop_state.load(Ordering::Acquire), STOP_STATE_TASK_RUNNING);
    }

    #[test]
    fn test_stop_fallback_needed_claims_reserved_state() {
        let stop_state = AtomicU8::new(STOP_STATE_TASK_RESERVED);
        assert!(stop_fallback_needed(&stop_state));
        assert_eq!(
            stop_state.load(Ordering::Acquire),
            STOP_STATE_FALLBACK_RESERVED
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_invoke_stop_typing_marks_completed_even_if_waiter_is_aborted() {
        let stop_typing_notify = Arc::new(Notify::new());
        let plugin = Arc::new(MockChannel::with_stop_typing_delay_and_notify(
            ChannelCapabilities {
                typing_indicators: true,
                ..Default::default()
            },
            Duration::from_millis(50),
            stop_typing_notify.clone(),
        ));
        let stop_state = Arc::new(AtomicU8::new(STOP_STATE_TASK_RESERVED));
        let task = tokio::spawn(invoke_stop_typing_with_running_state(
            plugin.clone(),
            TypingContext {
                to: "+15551234567".to_string(),
                ..Default::default()
            },
            stop_state.clone(),
        ));

        tokio::time::timeout(Duration::from_secs(1), async {
            while stop_state.load(Ordering::Acquire) != STOP_STATE_TASK_RUNNING {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("stop worker should enter TASK_RUNNING");

        task.abort();
        tokio::time::timeout(Duration::from_secs(1), stop_typing_notify.notified())
            .await
            .expect("aborting the waiter should not cancel a committed stop worker");
        tokio::time::timeout(Duration::from_secs(1), async {
            while stop_state.load(Ordering::Acquire) != STOP_STATE_COMPLETED {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("committed stop worker should still mark completion after waiter abort");

        assert_eq!(plugin.stop_typing_count.load(Ordering::Relaxed), 1);
        assert_eq!(stop_state.load(Ordering::Acquire), STOP_STATE_COMPLETED);
    }

    #[test]
    fn test_spawn_stop_typing_worker_requires_reserved_state() {
        let plugin = Arc::new(MockChannel::new(ChannelCapabilities {
            typing_indicators: true,
            ..Default::default()
        }));
        let stop_state = Arc::new(AtomicU8::new(STOP_STATE_COMPLETED));
        let worker = spawn_stop_typing_worker(
            plugin,
            TypingContext {
                to: "+15551234567".to_string(),
                ..Default::default()
            },
            stop_state.clone(),
        );

        assert!(worker.is_none());
        assert_eq!(stop_state.load(Ordering::Acquire), STOP_STATE_COMPLETED);
    }

    #[test]
    fn test_stop_fallback_needed_preserves_completed_state() {
        let stop_state = AtomicU8::new(STOP_STATE_COMPLETED);
        assert!(!stop_fallback_needed(&stop_state));
        assert_eq!(stop_state.load(Ordering::Acquire), STOP_STATE_COMPLETED);
    }

    #[test]
    fn test_stop_fallback_needed_preserves_fallback_reserved_state() {
        let stop_state = AtomicU8::new(STOP_STATE_FALLBACK_RESERVED);
        assert!(!stop_fallback_needed(&stop_state));
        assert_eq!(
            stop_state.load(Ordering::Acquire),
            STOP_STATE_FALLBACK_RESERVED
        );
    }

    #[test]
    fn test_stop_fallback_needed_claims_task_spawning_state() {
        let stop_state = AtomicU8::new(STOP_STATE_TASK_SPAWNING);
        assert!(stop_fallback_needed(&stop_state));
        assert_eq!(
            stop_state.load(Ordering::Acquire),
            STOP_STATE_FALLBACK_RESERVED
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn test_reserved_stop_state_race_has_single_owner() {
        let plugin = Arc::new(MockChannel::new(ChannelCapabilities {
            typing_indicators: true,
            ..Default::default()
        }));
        let stop_state = Arc::new(AtomicU8::new(STOP_STATE_TASK_RESERVED));
        let barrier = Arc::new(tokio::sync::Barrier::new(2));

        let worker_state = stop_state.clone();
        let worker_barrier = barrier.clone();
        let worker_plugin = plugin.clone();
        let worker = tokio::spawn(async move {
            worker_barrier.wait().await;
            spawn_stop_typing_worker(
                worker_plugin,
                TypingContext {
                    to: "+15551234567".to_string(),
                    ..Default::default()
                },
                worker_state,
            )
        });

        let fallback_state = stop_state.clone();
        let fallback_barrier = barrier.clone();
        let fallback = tokio::spawn(async move {
            fallback_barrier.wait().await;
            stop_fallback_needed(fallback_state.as_ref())
        });

        let maybe_worker = worker.await.expect("worker claimant should return");
        let fallback_claimed = fallback.await.expect("fallback claimant should return");

        match (maybe_worker, fallback_claimed) {
            (Some(worker), false) => {
                worker
                    .await
                    .expect("stop worker should complete")
                    .expect("stop worker should succeed");
                assert_eq!(plugin.stop_typing_count.load(Ordering::Relaxed), 1);
                assert_eq!(stop_state.load(Ordering::Acquire), STOP_STATE_COMPLETED);
            }
            (Some(worker), true) => {
                worker
                    .await
                    .expect("detached stop worker should still join cleanly")
                    .expect("worker should no-op cleanly when fallback steals ownership");
                assert_eq!(
                    plugin.stop_typing_count.load(Ordering::Relaxed),
                    0,
                    "fallback ownership should prevent the detached worker from double-sending"
                );
                assert_eq!(
                    stop_state.load(Ordering::Acquire),
                    STOP_STATE_FALLBACK_RESERVED
                );
            }
            (None, true) => {
                assert_eq!(
                    stop_state.load(Ordering::Acquire),
                    STOP_STATE_FALLBACK_RESERVED
                );
            }
            (None, false) => {
                panic!("either the detached worker or fallback cleanup must claim ownership");
            }
        }
    }

    #[test]
    fn test_typing_refresh_retry_delay_backs_off_and_caps() {
        let base = Duration::from_secs(2);
        assert_eq!(typing_refresh_retry_delay(base, 0), Duration::from_secs(2));
        assert_eq!(typing_refresh_retry_delay(base, 1), Duration::from_secs(2));
        assert_eq!(typing_refresh_retry_delay(base, 2), Duration::from_secs(4));
        assert_eq!(typing_refresh_retry_delay(base, 3), Duration::from_secs(8));
        assert_eq!(
            typing_refresh_retry_delay(base, 10),
            Duration::from_secs(30)
        );
    }

    #[test]
    fn test_next_typing_refresh_deadline_stays_on_wall_clock_cadence() {
        let start = tokio::time::Instant::now();
        let cadence = Duration::from_secs(3);

        let next =
            next_typing_refresh_deadline(start + cadence, cadence, start + Duration::from_secs(7));

        assert_eq!(next, start + Duration::from_secs(9));
    }

    #[test]
    fn test_should_log_typing_refresh_failure_throttles_noise() {
        assert!(should_log_typing_refresh_failure(1));
        assert!(should_log_typing_refresh_failure(2));
        assert!(should_log_typing_refresh_failure(3));
        assert!(should_log_typing_refresh_failure(4));
        assert!(!should_log_typing_refresh_failure(5));
        assert!(should_log_typing_refresh_failure(8));
    }

    #[test]
    fn test_activity_dispatcher_try_with_options_reports_thread_spawn_error() {
        fn fail_spawner(
            _builder: thread::Builder,
            routine: crate::thread_util::NamedThreadRoutine,
        ) -> io::Result<thread::JoinHandle<()>> {
            drop(routine);
            Err(io::Error::other("simulated stop typing thread exhaustion"))
        }

        let err = match ActivityDispatcher::try_with_options_and_spawner_for_test(1, fail_spawner) {
            Ok(_) => panic!("dispatcher startup should report thread spawn failure"),
            Err(err) => err,
        };

        let ActivityStartupError::ThreadSpawn { source: err } = err;
        let io_source = err
            .source()
            .expect("thread spawn error should preserve the original io::Error source");
        let io_error = io_source
            .downcast_ref::<io::Error>()
            .expect("thread spawn error source should remain an io::Error");
        assert_eq!(io_error.kind(), io::ErrorKind::Other);
        assert!(err.to_string().contains("failed to spawn startup thread"));
    }

    #[test]
    fn test_stop_typing_thread_spawn_error_preserves_source_error() {
        let err =
            StartupThreadSpawnError::new(STOP_TYPING_THREAD_NAME, io::Error::from_raw_os_error(11));

        let io_source = err
            .source()
            .expect("thread spawn error should preserve the original io::Error source");
        let io_error = io_source
            .downcast_ref::<io::Error>()
            .expect("thread spawn error source should remain an io::Error");

        assert_eq!(err.thread_name(), STOP_TYPING_THREAD_NAME);
        assert_eq!(io_error.raw_os_error(), Some(11));
    }

    #[test]
    fn test_stop_typing_dispatcher_does_not_drop_bursty_distinct_cleanup_requests() {
        let dispatcher = ActivityDispatcher::with_backlog_warning_threshold(1);
        let plugin = Arc::new(MockChannel::with_stop_typing_delay(
            ChannelCapabilities {
                typing_indicators: true,
                ..Default::default()
            },
            Duration::from_millis(10),
        ));
        let stop_states = (0..32_u32)
            .map(|recipient_index| {
                (
                    recipient_index,
                    Arc::new(AtomicU8::new(STOP_STATE_FALLBACK_RESERVED)),
                )
            })
            .collect::<Vec<_>>();

        for (recipient_index, stop_state) in &stop_states {
            dispatcher.dispatch_stop_typing(
                plugin.clone(),
                "signal",
                TypingContext {
                    to: format!("+1555123{recipient_index:04}"),
                    ..Default::default()
                },
                stop_state.clone(),
            );
        }

        std::thread::sleep(Duration::from_millis(500));
        dispatcher.shutdown();

        assert_eq!(plugin.stop_typing_count.load(Ordering::Relaxed), 32);
        assert!(stop_states
            .iter()
            .all(|(_, state)| state.load(Ordering::Acquire) == STOP_STATE_COMPLETED));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_stop_typing_dispatcher_coalesces_duplicate_recipient_cleanup_requests() {
        let stop_typing_started_notify = Arc::new(Notify::new());
        let dispatcher = ActivityDispatcher::with_backlog_warning_threshold(1);
        let plugin = Arc::new(MockChannel::with_stop_typing_delay_and_started_notify(
            ChannelCapabilities {
                typing_indicators: true,
                ..Default::default()
            },
            Duration::from_millis(50),
            stop_typing_started_notify.clone(),
        ));

        let first_stop_state = Arc::new(AtomicU8::new(STOP_STATE_FALLBACK_RESERVED));
        dispatcher.dispatch_stop_typing(
            plugin.clone(),
            "signal",
            TypingContext {
                to: "+15551234567".to_string(),
                ..Default::default()
            },
            first_stop_state.clone(),
        );

        tokio::time::timeout(
            Duration::from_secs(1),
            stop_typing_started_notify.notified(),
        )
        .await
        .expect("first stop_typing call should start");

        let additional_stop_states = (0..8)
            .map(|_| Arc::new(AtomicU8::new(STOP_STATE_FALLBACK_RESERVED)))
            .collect::<Vec<_>>();

        for stop_state in &additional_stop_states {
            dispatcher.dispatch_stop_typing(
                plugin.clone(),
                "signal",
                TypingContext {
                    to: "+15551234567".to_string(),
                    ..Default::default()
                },
                stop_state.clone(),
            );
        }

        dispatcher.shutdown();

        assert_eq!(
            plugin.stop_typing_count.load(Ordering::Relaxed),
            2,
            "duplicate recipient cleanup should coalesce into one in-flight call plus one rerun"
        );
        assert_eq!(
            first_stop_state.load(Ordering::Acquire),
            STOP_STATE_COMPLETED
        );
        assert!(additional_stop_states
            .iter()
            .all(|state| state.load(Ordering::Acquire) == STOP_STATE_COMPLETED));
    }

    #[tokio::test]
    async fn test_send_verified_read_receipt_marks_read() {
        let plugin = Arc::new(MockChannel::new(ChannelCapabilities {
            read_receipts: true,
            ..Default::default()
        }));
        send_verified_read_receipt_with_plugin(
            plugin.clone(),
            "signal",
            ReadReceiptContext {
                recipient: "+15551234567".to_string(),
                timestamp: Some(123),
                ..Default::default()
            },
        )
        .await
        .expect("read receipt dispatch should succeed");

        assert_eq!(plugin.mark_read_count.load(Ordering::Relaxed), 1);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_send_verified_read_receipt_reports_plugin_panic_as_permanent_failure() {
        let plugin = Arc::new(MockChannel {
            ..MockChannel::with_panicking_mark_read(
                ChannelCapabilities {
                    read_receipts: true,
                    ..Default::default()
                },
                1,
            )
        });

        let err = send_verified_read_receipt_with_plugin(
            plugin,
            "signal",
            ReadReceiptContext {
                recipient: "+15551234567".to_string(),
                timestamp: Some(123),
                ..Default::default()
            },
        )
        .await
        .expect_err("plugin panic should report a permanent dispatch failure");

        assert!(matches!(err, ReadReceiptDispatchError::Permanent(_)));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_send_verified_read_receipt_waits_for_slow_io() {
        let plugin = Arc::new(MockChannel::with_mark_read_delay(
            ChannelCapabilities {
                read_receipts: true,
                ..Default::default()
            },
            Duration::from_millis(150),
        ));

        let started_at = std::time::Instant::now();
        send_verified_read_receipt_with_plugin(
            plugin.clone(),
            "signal",
            ReadReceiptContext {
                recipient: "+15551234567".to_string(),
                timestamp: Some(1),
                ..Default::default()
            },
        )
        .await
        .expect("slow read receipt should still succeed");

        assert!(
            started_at.elapsed() >= Duration::from_millis(100),
            "direct read receipt dispatch should reflect the underlying slow I/O"
        );
        assert_eq!(plugin.mark_read_count.load(Ordering::Relaxed), 1);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_read_receipt_task_executor_fails_when_capability_disappears_after_activation() {
        let plugin_registry = Arc::new(PluginRegistry::new());
        plugin_registry.register_channel(
            "signal".to_string(),
            Arc::new(MockChannel::new(ChannelCapabilities::default())),
        );
        let state = Arc::new(
            crate::server::ws::WsServerState::new(crate::server::ws::WsServerConfig::default())
                .with_plugin_registry(plugin_registry),
        );
        let executor = ReadReceiptTaskExecutor {
            state: state.clone(),
        };

        let task = state
            .activity_service()
            .read_receipt_queue()
            .enqueue_async(
                serde_json::to_value(ReadReceiptTaskPayload::new(
                    "signal",
                    ReadReceiptContext {
                        recipient: "+15551234567".to_string(),
                        timestamp: Some(123),
                        ..Default::default()
                    },
                ))
                .expect("read receipt payload should serialize"),
                None,
            )
            .await;
        let claimed = state
            .activity_service()
            .read_receipt_queue()
            .claim_due(crate::time::unix_now_ms_u64(), 1);
        assert_eq!(claimed.len(), 1);
        assert_eq!(claimed[0].id, task.id);

        let outcome = executor.execute(claimed[0].clone()).await;

        match outcome {
            TaskExecutionOutcome::Failed { error } => {
                assert!(
                    error.contains("does not support read receipts after activation"),
                    "unexpected failure message: {error}"
                );
            }
            other => panic!("expected terminal failure, got {other:?}"),
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_read_receipt_task_executor_rejects_unsupported_kind() {
        let state = Arc::new(crate::server::ws::WsServerState::new(
            crate::server::ws::WsServerConfig::default(),
        ));
        let executor = ReadReceiptTaskExecutor {
            state: state.clone(),
        };

        let task = state
            .activity_service()
            .read_receipt_queue()
            .enqueue_async(
                serde_json::json!({
                    "kind": "activityReadReceipt",
                    "channelId": "signal",
                    "context": {
                        "recipient": "+15551234567",
                        "timestamp": 123
                    }
                }),
                None,
            )
            .await;

        let claimed = state
            .activity_service()
            .read_receipt_queue()
            .claim_due(crate::time::unix_now_ms_u64(), 1);
        assert_eq!(claimed.len(), 1);
        assert_eq!(claimed[0].id, task.id);

        let outcome = executor.execute(claimed[0].clone()).await;
        match outcome {
            TaskExecutionOutcome::Failed { error } => {
                assert!(
                    error.contains("unsupported read receipt task kind"),
                    "{error}"
                );
                assert!(error.contains("activity/read_receipts.json"), "{error}");
            }
            other => panic!("expected unsupported kind failure, got {other:?}"),
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_read_receipt_task_executor_respects_max_attempts_budget() {
        let state = Arc::new(crate::server::ws::WsServerState::new(
            crate::server::ws::WsServerConfig::default(),
        ));
        let executor = ReadReceiptTaskExecutor {
            state: state.clone(),
        };

        let task = state
            .activity_service()
            .read_receipt_queue()
            .enqueue_async_with_policy(
                serde_json::to_value(ReadReceiptTaskPayload::new(
                    "signal",
                    ReadReceiptContext {
                        recipient: "+15551234567".to_string(),
                        timestamp: Some(123),
                        ..Default::default()
                    },
                ))
                .expect("read receipt payload should serialize"),
                None,
                crate::tasks::TaskPolicy {
                    max_attempts: 1,
                    ..Default::default()
                },
            )
            .await;

        let first_claimed = state
            .activity_service()
            .read_receipt_queue()
            .claim_due(crate::time::unix_now_ms_u64(), 1);
        assert_eq!(first_claimed.len(), 1);
        assert_eq!(first_claimed[0].id, task.id);
        assert_eq!(first_claimed[0].attempts, 1);

        let first_error = match executor.execute(first_claimed[0].clone()).await {
            TaskExecutionOutcome::RetryWait { delay_ms, error } => {
                assert_eq!(delay_ms, READ_RECEIPT_RETRY_DELAY_MS);
                error
            }
            other => panic!("expected retryable first attempt, got {other:?}"),
        };
        assert!(state
            .activity_service()
            .read_receipt_queue()
            .mark_retry_wait(&task.id, 0, &first_error));

        let second_claimed = state
            .activity_service()
            .read_receipt_queue()
            .claim_due(crate::time::unix_now_ms_u64(), 1);
        assert_eq!(second_claimed.len(), 1);
        assert_eq!(second_claimed[0].id, task.id);
        assert_eq!(second_claimed[0].attempts, 2);

        assert_eq!(
            executor.execute(second_claimed[0].clone()).await,
            TaskExecutionOutcome::Failed {
                error: "objective policy violation: attempts 2 exceeded maxAttempts 1".to_string(),
            }
        );
    }

    #[test]
    fn test_reset_unsupported_activity_feature_warnings_for_test_clears_seen_keys() {
        let service = ActivityService::new();
        service.reset_unsupported_activity_feature_warnings_for_test();
        service.warn_unsupported_feature("signal", "typing");
        assert!(service
            .unsupported_feature_warnings
            .lock()
            .seen_at
            .contains_key("signal:typing"));

        service.reset_unsupported_activity_feature_warnings_for_test();
        assert!(service
            .unsupported_feature_warnings
            .lock()
            .seen_at
            .is_empty());
    }

    #[test]
    fn test_unsupported_activity_warning_registry_rewarns_after_cooldown() {
        let mut registry =
            UnsupportedActivityWarningRegistry::with_cooldown_for_test(Duration::from_secs(1));
        let start = Instant::now();

        assert!(registry.should_warn("signal:typing", start));
        assert!(!registry.should_warn("signal:typing", start + Duration::from_millis(500)));
        assert!(registry.should_warn("signal:typing", start + Duration::from_secs(2)));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_collect_configured_unsupported_features_for_registered_channels() {
        let _fixture = crate::test_support::config::StableConfigFixture::new(serde_json::json!({
            "channels": {
                "custom": {
                    "features": {
                        "typing": { "enabled": true },
                        "readReceipts": { "enabled": true },
                    }
                }
            }
        }));

        let plugin_registry = Arc::new(PluginRegistry::new());
        plugin_registry.register_channel(
            "custom".to_string(),
            Arc::new(MockChannel::new(ChannelCapabilities::default())),
        );

        let unsupported =
            collect_configured_unsupported_features_for_registered_channels(plugin_registry).await;

        assert!(unsupported.contains(&("custom".to_string(), "typing")));
        assert!(unsupported.contains(&("custom".to_string(), "read_receipts")));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_can_accept_read_receipt_ownership_disables_new_obligations_at_high_watermark() {
        let service = ActivityService::with_limits_for_test(8, 2);
        assert!(service.can_accept_read_receipt_ownership("signal"));

        let first = service
            .enqueue_ready_read_receipt(
                "signal",
                ReadReceiptContext {
                    recipient: "+15551230001".to_string(),
                    timestamp: Some(1),
                    ..Default::default()
                },
            )
            .await;
        let second = service
            .enqueue_ready_read_receipt(
                "signal",
                ReadReceiptContext {
                    recipient: "+15551230002".to_string(),
                    timestamp: Some(2),
                    ..Default::default()
                },
            )
            .await;

        assert!(first.is_some());
        assert!(second.is_some());
        assert!(
            !service.can_accept_read_receipt_ownership("signal"),
            "new ownership should stop once the durable backlog reaches the high-water mark"
        );
    }

    #[test]
    fn test_try_claim_read_receipt_counts_and_releases_receive_time_reservations() {
        let service = ActivityService::with_limits_for_test(8, 1);
        let claim = service
            .try_claim_read_receipt(
                "signal",
                ReadReceiptContext {
                    recipient: "+15551230001".to_string(),
                    timestamp: Some(1),
                    ..Default::default()
                },
            )
            .expect("first claim should reserve the only available receipt slot");
        assert!(
            !service.can_accept_read_receipt_ownership("signal"),
            "tracked receive-time claims should count against the ownership high-water mark"
        );

        assert!(service.withhold_claimed_read_receipt(&claim));
        assert!(
            service.can_accept_read_receipt_ownership("signal"),
            "withholding a tracked claim should release its ownership reservation"
        );
    }

    #[test]
    fn test_poll_reservation_releases_unused_capacity_on_drop() {
        let service = ActivityService::with_limits_for_test(8, 2);
        let mut reservation = service
            .reserve_available_read_receipt_ownership("signal")
            .expect("poll reservation should reserve the available ownership capacity");
        assert_eq!(reservation.reserved_capacity(), 2);
        assert!(
            !service.can_accept_read_receipt_ownership("signal"),
            "reserved poll capacity should count against the ownership high-water mark"
        );

        let claimed = reservation
            .claim(ReadReceiptContext {
                recipient: "+15551230001".to_string(),
                timestamp: Some(1),
                ..Default::default()
            })
            .expect("claim should consume one reserved slot");
        assert_eq!(reservation.reserved_capacity(), 1);

        drop(reservation);
        assert!(
            service.can_accept_read_receipt_ownership("signal"),
            "dropping the reservation should release unused poll slots immediately while keeping claimed slots reserved"
        );

        assert!(service.withhold_claimed_read_receipt(&claimed));
        assert!(
            service.can_accept_read_receipt_ownership("signal"),
            "releasing the claimed slot should restore the now-unused reserved capacity"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_complete_claimed_read_receipt_releases_capacity_before_direct_send_finishes() {
        let mark_read_started = Arc::new(Notify::new());
        let plugin = Arc::new(MockChannel {
            caps: ChannelCapabilities {
                read_receipts: true,
                ..Default::default()
            },
            mark_read_started_notify: Some(mark_read_started.clone()),
            mark_read_delay: Duration::from_millis(150),
            ..MockChannel::new(ChannelCapabilities {
                read_receipts: true,
                ..Default::default()
            })
        });
        let plugin_registry = Arc::new(PluginRegistry::new());
        plugin_registry.register_channel("signal".to_string(), plugin.clone());
        let service = Arc::new(
            ActivityService::with_read_receipt_queue_and_limits_for_test(
                Arc::new(TaskQueue::with_capacity_limit(None, Some(0))),
                8,
                1,
            ),
        );
        let state = Arc::new(
            crate::server::ws::WsServerState::new(crate::server::ws::WsServerConfig::default())
                .with_plugin_registry(plugin_registry)
                .with_activity_service(service.clone()),
        );

        let first_claim = service
            .try_claim_read_receipt(
                "signal",
                ReadReceiptContext {
                    recipient: "+15551230001".to_string(),
                    timestamp: Some(1),
                    ..Default::default()
                },
            )
            .expect("first claim should reserve the only ownership slot");
        assert!(
            !service.can_accept_read_receipt_ownership("signal"),
            "the claimed slot should consume the full high-watermark budget"
        );

        let completion = tokio::spawn({
            let service = service.clone();
            let state = state.clone();
            let first_claim = first_claim.clone();
            async move {
                service
                    .complete_claimed_read_receipt(state.as_ref(), &first_claim)
                    .await
            }
        });

        tokio::time::timeout(Duration::from_secs(1), mark_read_started.notified())
            .await
            .expect("direct-send fallback should enter mark_read");

        let second_claim = service
            .try_claim_read_receipt(
                "signal",
                ReadReceiptContext {
                    recipient: "+15551230002".to_string(),
                    timestamp: Some(2),
                    ..Default::default()
                },
            )
            .expect("capacity should be released before the direct send finishes");
        assert!(service.withhold_claimed_read_receipt(&second_claim));

        completion
            .await
            .expect("completion task should join cleanly")
            .expect("direct-send fallback should succeed");
        assert_eq!(plugin.mark_read_count.load(Ordering::Relaxed), 1);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_acknowledge_read_receipt_now_returns_err_when_retry_is_scheduled() {
        let service = Arc::new(ActivityService::new());
        let state = Arc::new(
            crate::server::ws::WsServerState::new(crate::server::ws::WsServerConfig::default())
                .with_activity_service(service.clone()),
        );

        let err = service
            .acknowledge_read_receipt_now(
                state.as_ref(),
                "signal",
                ReadReceiptContext {
                    recipient: "+15551234567".to_string(),
                    timestamp: Some(123),
                    ..Default::default()
                },
            )
            .await
            .expect_err("retryable inline dispatch should not report success");

        assert!(
            err.contains("plugin registry unavailable"),
            "retryable error should surface the underlying dispatch reason"
        );
        let tasks = service.read_receipt_queue().list();
        assert_eq!(tasks.len(), 1);
        assert_eq!(tasks[0].state, crate::tasks::TaskState::RetryWait);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn test_acknowledge_read_receipt_now_falls_back_to_direct_send_when_persist_fails() {
        let plugin = Arc::new(MockChannel::new(ChannelCapabilities {
            read_receipts: true,
            ..Default::default()
        }));
        let plugin_registry = Arc::new(PluginRegistry::new());
        plugin_registry.register_channel("signal".to_string(), plugin.clone());
        let service = Arc::new(ActivityService::with_read_receipt_queue_for_test(Arc::new(
            TaskQueue::with_capacity_limit(None, Some(0)),
        )));
        let state = Arc::new(
            crate::server::ws::WsServerState::new(crate::server::ws::WsServerConfig::default())
                .with_plugin_registry(plugin_registry)
                .with_activity_service(service.clone()),
        );

        service
            .acknowledge_read_receipt_now(
                state.as_ref(),
                "signal",
                ReadReceiptContext {
                    recipient: "+15551234567".to_string(),
                    timestamp: Some(123),
                    ..Default::default()
                },
            )
            .await
            .expect("persist failure should fall back to direct read receipt dispatch");

        assert_eq!(plugin.mark_read_count.load(Ordering::Relaxed), 1);
        assert!(
            service.read_receipt_queue().list().is_empty(),
            "synthetic failed enqueue should not leave a persisted queue entry"
        );
    }

    #[test]
    fn test_inline_read_receipt_result_reports_sent_but_unfinalized_done_state() {
        let result = ActivityService::inline_read_receipt_result(
            TaskExecutionOutcome::Done { run_id: None },
            Err("queue update failed".to_string()),
        );

        let err =
            result.expect_err("done outcome with failed finalization should surface an error");
        assert!(
            err.contains("read receipt was sent but durable task finalization failed"),
            "unexpected error message: {err}"
        );
        assert!(
            err.contains("queue update failed"),
            "unexpected error message: {err}"
        );
    }
}
