futu_mcp/state/

push_subscribers.rs

use std::collections::{HashSet, VecDeque};
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::sync::{Arc, Mutex as StdMutex, Weak};
use std::time::Instant;

use rmcp::{
    RoleServer,
    model::Resource,
    service::{Peer, SubscriptionSink},
};
use tokio::sync::{OwnedSemaphorePermit, TryAcquireError};

use super::ServerState;
use super::push_filter::subscriber_visible_to_caller;

mod modern_delivery;
mod visibility;
use visibility::current_scope_allows_subscriber;
pub(crate) use visibility::{parse_push_resource_uri, push_resource_uri};

pub(crate) const PUSH_RESOURCE_PREFIX: &str = "futu://push/";
pub(crate) const MODERN_PUSH_QUEUE_CAPACITY: usize = 64;
pub(crate) const MAX_MODERN_PUSH_HANDLES: usize = 128;
pub(crate) const PUSH_SUBSCRIBER_MAX_AGE: std::time::Duration =
    std::time::Duration::from_secs(4 * 3600);

/// One rmcp service instance owns all legacy push rows created by that logical
/// transport session. Rows keep only a weak opaque identity, so TTL and
/// explicit unsubscribe do not leave a second, unbounded handle index behind.
pub(crate) struct LegacyPushServiceLease {
    state: ServerState,
    owner: Arc<()>,
    registration_attempted: AtomicBool,
}

impl LegacyPushServiceLease {
    pub(crate) fn new(state: ServerState) -> Self {
        Self {
            state,
            owner: Arc::new(()),
            registration_attempted: AtomicBool::new(false),
        }
    }

    pub(crate) async fn register_push_subscriber(
        &self,
        delivery: PushDeliveryTarget,
        acc_ids: HashSet<u64>,
        owner_key_id: Option<String>,
        allowed_acc_ids_snapshot: Option<HashSet<u64>>,
        allowed_markets_snapshot: Option<HashSet<String>>,
    ) -> Result<String, String> {
        if !matches!(delivery, PushDeliveryTarget::LegacyPeer(_, _)) {
            return Err("legacy push service lease cannot own a modern resource".to_string());
        }
        // Store before the only await. A cancelled/failed registration may
        // cause one harmless empty scan, but can never leave an inserted row
        // without arming final service cleanup.
        self.registration_attempted.store(true, Ordering::Release);
        self.state
            .register_push_subscriber_with_owner(
                delivery,
                acc_ids,
                owner_key_id,
                allowed_acc_ids_snapshot,
                allowed_markets_snapshot,
                Some(Arc::downgrade(&self.owner)),
            )
            .await
    }

    #[cfg(test)]
    pub(crate) fn mark_registration_attempted_for_test(&self) {
        self.registration_attempted.store(true, Ordering::Release);
    }
}

impl Drop for LegacyPushServiceLease {
    fn drop(&mut self) {
        if !self.registration_attempted.load(Ordering::Acquire) {
            return;
        }
        // Pointer identity remains comparable after the last strong owner is
        // dropped. Do not use `upgrade()` in the cleanup path.
        let owner = Arc::downgrade(&self.owner);
        let state = self.state.clone();
        match tokio::runtime::Handle::try_current() {
            Ok(runtime) => {
                runtime.spawn(async move {
                    let removed = state
                        .remove_legacy_push_subscribers_for_service(&owner)
                        .await;
                    tracing::debug!(
                        removed,
                        "removed legacy MCP push subscribers after service close"
                    );
                });
            }
            Err(error) => {
                // Runtime shutdown is not evidence that async cleanup ran; the
                // existing subscriber TTL remains the fallback. Never panic in Drop.
                tracing::debug!(
                    %error,
                    "legacy MCP service dropped without a live Tokio runtime; deferred cleanup not scheduled"
                );
            }
        }
    }
}

#[derive(Clone)]
pub(crate) enum PushDeliveryTarget {
    LegacyPeer(Peer<RoleServer>, Arc<AtomicU8>),
    ModernResource(Arc<OwnedSemaphorePermit>),
}

#[derive(Clone)]
pub(super) enum SubscriberDelivery {
    LegacyPeer(Peer<RoleServer>, Arc<AtomicU8>),
    ModernResource(Arc<ModernResourceDelivery>),
}

struct ActiveModernListener {
    token: u64,
    sink: SubscriptionSink,
    cancel: tokio_util::sync::CancellationToken,
}

struct ModernResourceInner {
    queue: VecDeque<serde_json::Value>,
    dropped_events: u64,
    enqueued_generation: u64,
    next_listener_token: u64,
    active_listener: Option<ActiveModernListener>,
    pending_notification: Option<u64>,
    #[cfg(test)]
    notification_send_pause: Option<(Arc<tokio::sync::Barrier>, Arc<tokio::sync::Barrier>)>,
    #[cfg(test)]
    notification_send_attempts: u64,
    #[cfg(test)]
    notification_send_failures: u64,
    closed: bool,
}

pub(super) struct ModernResourceDelivery {
    handle: String,
    inner: StdMutex<ModernResourceInner>,
    _permit: Arc<OwnedSemaphorePermit>,
}

pub(super) struct PendingModernNotification {
    listener_token: u64,
    sink: SubscriptionSink,
    cancel: tokio_util::sync::CancellationToken,
    uri: String,
    generation: u64,
}

/// v1.4.38 Phase 5: 订阅了 push 通知的 MCP 客户端 session 记录。
///
/// 每个 session 用 `futu_sub_acc_push` 注册时登记一条。daemon push 事件来到
/// MCP server 时，按 `acc_ids` 过滤后用 `peer.notify_logging_message()` 推回。
///
/// v1.4.38 100%：`acc_ids` 过滤已生效（state.rs drain loop 实装）。
/// caller key ownership / scope 快照在注册时解析并保存，后续 push 分发不再
/// 重新读取 bearer 明文。
#[derive(Clone)]
pub(super) struct PushSubscriber {
    pub delivery: SubscriberDelivery,
    /// 该 session 关心的 acc_id 列表。空集合表示"不过滤"（接收所有 acc 的 push）。
    pub acc_ids: HashSet<u64>,
    /// v1.4.39 per-key acc_id 白名单**注册时快照**（非 live-reload）。
    ///
    /// Some(set) + non-empty → push 的 acc_id 必须在 set 里才推。
    /// Some(empty) / None → 不做 key 级过滤（兼容无 allowed_acc_ids 约束的 key
    /// 或 stdio / legacy 模式）。
    ///
    /// 快照语义：注册后 SIGHUP 重载 keys.json 修改 allowed_acc_ids 不会立即
    /// 反映到已注册订阅者。用户需重新 `futu_sub_acc_push` 才应用新 scope。
    /// 这是 defense-in-depth 层（主 auth 在 tool 调用时 guard.rs），可接受。
    pub allowed_acc_ids_snapshot: Option<HashSet<u64>>,
    /// v1.4.105 T-C2: per-key `allowed_markets` 注册时快照, Layer 3 trade push gate.
    ///
    /// `Some(set)` + non-empty → push event 的 `trd_market` 必须 ∈ set 才推 (按
    /// `Trd_Common.TrdMarket` int → 字符串映射, 与 `keys.json::allowed_markets`
    /// 配置字符串一致, e.g. "HK"/"US"/"FUTURES").
    /// `None` / `Some(empty)` → 不做 market gate (兼容 stdio / legacy / 未配
    /// allowed_markets 的 key).
    ///
    /// 与 `allowed_acc_ids_snapshot` 同样**注册时快照**, SIGHUP 重载不影响
    /// 已注册订阅者. 用户需重新 `futu_sub_acc_push` 才应用新 scope.
    /// 配套 main auth (guard.rs / require_acc_read_with_acc_id) 仍在 tool 调用
    /// 时 enforce, 这是 defense-in-depth 层 (push 走 server-initiated channel
    /// 绕过 tool 调用 → 必须独立 enforce).
    pub allowed_markets_snapshot: Option<HashSet<String>>,
    /// 注册时间（用于 session 硬上限清理，4h 默认 TTL）
    pub registered_at: Instant,
    /// v1.4.103 (codex 50 F6 / 53 F4 — B8): owner key id (KeyRecord.id).
    /// 注册时填的 caller key id (HTTP Bearer 或 startup key); 用于 unsub
    /// ownership check — 任何 caller 拿到 session_id 后想 unsub 必须 key id
    /// 匹配 owner_key_id (admin scope 例外).
    ///
    /// None = legacy / stdio 模式无 key (ownership 退化为 "anyone can unsub",
    /// 与本来 v1.4.102 行为一致).
    pub owner_key_id: Option<String>,
    /// Opaque owner of a legacy rmcp service/session. Modern resources are
    /// always `None` and must never be removed by legacy session cleanup.
    pub legacy_service_owner: Option<Weak<()>>,
}

impl ServerState {
    pub async fn reserve_modern_push_handle(&self) -> Result<Arc<OwnedSemaphorePermit>, String> {
        self.modern_push_slots
            .clone()
            .try_acquire_owned()
            .map(Arc::new)
            .map_err(|error| match error {
                TryAcquireError::NoPermits => format!(
                    "modern push handle limit reached ({MAX_MODERN_PUSH_HANDLES}); unsubscribe an existing handle before retrying"
                ),
                TryAcquireError::Closed => "modern push handle registry is closed".to_string(),
            })
    }

    /// v1.4.38 Phase 5: 注册当前 session 接收指定 acc_id 的 push。返回 session
    /// UUID（调用方存着，后续可 unregister）。
    ///
    /// v1.4.38: 已 wire 到 `futu_sub_acc_push` tool。tool 被调用时拿到
    /// `RequestContext.peer`，`acc_ids` 从工具 args 解析，注册完成后
    /// state.rs 的 push drain loop 会按 acc_ids filter 转 notify 给该 peer。
    /// v1.4.103 (codex 50 F5 / 53 F2 / 58 F3 — B7) + (codex 50 F6 / 53 F4 — B8):
    /// 注册当前 session 接收指定 acc_id 的 push。
    ///
    /// Authorization fields are supplied from one already-approved immutable
    /// caller snapshot. This method never re-resolves plaintext after a daemon
    /// await, avoiding identity/scope drift between authorization and insert.
    pub async fn register_push_subscriber_with_owner(
        &self,
        delivery: PushDeliveryTarget,
        acc_ids: HashSet<u64>,
        owner_key_id: Option<String>,
        allowed_acc_ids_snapshot: Option<HashSet<u64>>,
        allowed_markets_snapshot: Option<HashSet<String>>,
        legacy_service_owner: Option<Weak<()>>,
    ) -> Result<String, String> {
        match (&delivery, &legacy_service_owner) {
            (PushDeliveryTarget::ModernResource(_), Some(_)) => {
                return Err("modern push resource cannot have a legacy service owner".to_string());
            }
            (PushDeliveryTarget::LegacyPeer(_, _), None) => {
                return Err("legacy push subscriber requires a service owner".to_string());
            }
            _ => {}
        }
        let mut subscribers = self.push_subscribers.lock().await;
        for _ in 0..16 {
            let session_id = format!("sub-{}", rand::random::<u128>());
            if subscribers.contains_key(&session_id) {
                continue;
            }
            let delivery = match delivery.clone() {
                PushDeliveryTarget::LegacyPeer(peer, minimum_level) => {
                    SubscriberDelivery::LegacyPeer(peer, minimum_level)
                }
                PushDeliveryTarget::ModernResource(permit) => SubscriberDelivery::ModernResource(
                    Arc::new(ModernResourceDelivery::new(session_id.clone(), permit)),
                ),
            };
            subscribers.insert(
                session_id.clone(),
                PushSubscriber {
                    delivery,
                    acc_ids: acc_ids.clone(),
                    allowed_acc_ids_snapshot: allowed_acc_ids_snapshot.clone(),
                    allowed_markets_snapshot: allowed_markets_snapshot.clone(),
                    registered_at: Instant::now(),
                    owner_key_id: owner_key_id.clone(),
                    legacy_service_owner: legacy_service_owner.clone(),
                },
            );
            return Ok(session_id);
        }
        Err("failed to allocate a collision-free push resource handle".to_string())
    }

    async fn remove_legacy_push_subscribers_for_service(&self, owner: &Weak<()>) -> usize {
        let mut subscribers = self.push_subscribers.lock().await;
        let handles = subscribers
            .iter()
            .filter_map(|(handle, subscriber)| {
                let is_same_legacy_service =
                    matches!(subscriber.delivery, SubscriberDelivery::LegacyPeer(_, _))
                        && subscriber
                            .legacy_service_owner
                            .as_ref()
                            .is_some_and(|candidate| Weak::ptr_eq(candidate, owner));
                is_same_legacy_service.then(|| handle.clone())
            })
            .collect::<Vec<_>>();
        for handle in &handles {
            subscribers.remove(handle);
        }
        handles.len()
    }

    /// v1.4.103 (codex 50 F6 / 53 F4 — B8): unsub session ownership check.
    ///
    /// 行为:
    /// - 无 caller_key_id (legacy / stdio): 退化为旧行为 (任何 caller 可 unsub).
    /// - 有 caller_key_id + subscriber.owner_key_id 匹配: 删除, 返 Ok(true).
    /// - 有 caller_key_id + subscriber.owner_key_id 不匹配: **拒绝**, 返
    ///   Err(reason) — 防其他 caller 拿可见 session_id 强制 unsub.
    /// - session_id 不存在: 返 Ok(false) (idempotent, 不报错).
    /// - subscriber.owner_key_id = None (legacy 注册): 退化为旧行为 — 任何 caller
    ///   可 unsub (向后兼容).
    pub async fn unregister_push_subscriber_with_owner_check(
        &self,
        session_id: &str,
        caller_key_id: Option<&str>,
    ) -> Result<bool, String> {
        self.purge_expired_push_subscribers().await;
        let mut subs = self.push_subscribers.lock().await;
        // 不存在 → idempotent Ok(false), 不报错
        let Some(sub) = subs.get(session_id) else {
            return Ok(false);
        };
        // ownership check
        match (caller_key_id, sub.owner_key_id.as_deref()) {
            (None, _) => {
                // 无 caller key: legacy / stdio 模式, 退化旧行为
                let removed = subs.remove(session_id);
                drop(subs);
                if let Some(subscriber) = removed {
                    subscriber.close_modern_resource();
                }
                Ok(true)
            }
            (Some(caller), None) => {
                // session 注册时无 owner_key_id (legacy): 任何 caller 可 unsub
                tracing::warn!(
                    session_id,
                    caller,
                    "v1.4.103 B8: unsub legacy session (no owner_key_id) — \
                     allowed for backward-compat"
                );
                let removed = subs.remove(session_id);
                drop(subs);
                if let Some(subscriber) = removed {
                    subscriber.close_modern_resource();
                }
                Ok(true)
            }
            (Some(caller), Some(owner)) if caller == owner => {
                let removed = subs.remove(session_id);
                drop(subs);
                if let Some(subscriber) = removed {
                    subscriber.close_modern_resource();
                }
                Ok(true)
            }
            (Some(caller), Some(owner)) => {
                // ownership 不匹配: reject. 当前 MCP subscription ownership contract
                // 没有 admin override surface；若要扩展，必须先在 spec / auth
                // pipeline / integration tests 中定义清楚。
                if matches!(sub.delivery, SubscriberDelivery::ModernResource(_)) {
                    Err("push resource not found for current caller".to_string())
                } else {
                    Err(format!(
                        "session_id {session_id:?} owned by key_id {owner:?}, \
                         caller key_id {caller:?} not allowed to unsub"
                    ))
                }
            }
        }
    }

    pub async fn modern_resources_for_caller(
        &self,
        caller_key_id: &str,
        caller_allowed_acc_ids: Option<&HashSet<u64>>,
        caller_allowed_markets: Option<&HashSet<String>>,
    ) -> Vec<Resource> {
        self.purge_expired_push_subscribers().await;
        let now = Instant::now();
        self.push_subscribers
            .lock()
            .await
            .iter()
            .filter(|(_, subscriber)| {
                matches!(subscriber.delivery, SubscriberDelivery::ModernResource(_))
                    && subscriber.owner_key_id.as_deref() == Some(caller_key_id)
                    && now
                        .checked_duration_since(subscriber.registered_at)
                        .is_none_or(|age| age < PUSH_SUBSCRIBER_MAX_AGE)
                    && current_scope_allows_subscriber(
                        subscriber,
                        caller_allowed_acc_ids,
                        caller_allowed_markets,
                    )
            })
            .map(|(handle, _)| {
                Resource::new(push_resource_uri(handle), "futu-account-push")
                    .with_title("Futu account push events")
                    .with_description("Bounded private queue of authorized Futu account pushes")
                    .with_mime_type("application/json")
            })
            .collect()
    }

    fn modern_resource_for_caller(
        subscriber: &PushSubscriber,
        caller_key_id: &str,
        caller_allowed_acc_ids: Option<&HashSet<u64>>,
        caller_allowed_markets: Option<&HashSet<String>>,
    ) -> Result<Arc<ModernResourceDelivery>, String> {
        if subscriber.owner_key_id.as_deref() != Some(caller_key_id)
            || !current_scope_allows_subscriber(
                subscriber,
                caller_allowed_acc_ids,
                caller_allowed_markets,
            )
            || Instant::now()
                .checked_duration_since(subscriber.registered_at)
                .is_some_and(|age| age >= PUSH_SUBSCRIBER_MAX_AGE)
        {
            return Err("push resource not found for current caller".to_string());
        }
        match &subscriber.delivery {
            SubscriberDelivery::ModernResource(resource) => Ok(resource.clone()),
            SubscriberDelivery::LegacyPeer(_, _) => {
                Err("push resource not found for current caller".to_string())
            }
        }
    }

    pub async fn drain_modern_push_resource(
        &self,
        uri: &str,
        caller_key_id: &str,
        caller_allowed_acc_ids: Option<&HashSet<u64>>,
        caller_allowed_markets: Option<&HashSet<String>>,
    ) -> Result<(String, Vec<serde_json::Value>, u64), String> {
        self.purge_expired_push_subscribers().await;
        let handle = parse_push_resource_uri(uri)
            .ok_or_else(|| "invalid futu push resource URI".to_string())?;
        let resource = {
            let subscribers = self.push_subscribers.lock().await;
            let subscriber = subscribers
                .get(handle)
                .ok_or_else(|| "push resource not found for current caller".to_string())?;
            Self::modern_resource_for_caller(
                subscriber,
                caller_key_id,
                caller_allowed_acc_ids,
                caller_allowed_markets,
            )?
        };
        let (events, dropped) = resource.drain()?;
        Ok((handle.to_string(), events, dropped))
    }

    pub async fn attach_modern_push_listener(
        &self,
        uri: &str,
        caller_key_id: &str,
        caller_allowed_acc_ids: Option<&HashSet<u64>>,
        caller_allowed_markets: Option<&HashSet<String>>,
        sink: SubscriptionSink,
    ) -> Result<(u64, tokio_util::sync::CancellationToken), String> {
        self.purge_expired_push_subscribers().await;
        let handle = parse_push_resource_uri(uri)
            .ok_or_else(|| "invalid futu push resource URI".to_string())?;
        let resource = {
            let subscribers = self.push_subscribers.lock().await;
            let subscriber = subscribers
                .get(handle)
                .ok_or_else(|| "push resource not found for current caller".to_string())?;
            Self::modern_resource_for_caller(
                subscriber,
                caller_key_id,
                caller_allowed_acc_ids,
                caller_allowed_markets,
            )?
        };
        let (listener_token, cancel, pending) = resource.attach_listener(sink)?;
        if let Some(work) = pending {
            let resource = Arc::clone(&resource);
            tokio::spawn(async move {
                resource.send_pending_notification(work).await;
            });
        }
        Ok((listener_token, cancel))
    }

    pub async fn detach_modern_push_listener(&self, uri: &str, listener_token: u64) {
        let Some(handle) = parse_push_resource_uri(uri) else {
            return;
        };
        let resource = self
            .push_subscribers
            .lock()
            .await
            .get(handle)
            .and_then(|subscriber| match &subscriber.delivery {
                SubscriberDelivery::ModernResource(resource) => Some(resource.clone()),
                SubscriberDelivery::LegacyPeer(_, _) => None,
            });
        if let Some(resource) = resource {
            resource.detach_listener_if(listener_token);
        }
    }

    pub(super) async fn purge_expired_push_subscribers(&self) -> usize {
        self.purge_expired_push_subscribers_at(Instant::now()).await
    }

    async fn purge_expired_push_subscribers_at(&self, now: Instant) -> usize {
        let removed = {
            let mut subscribers = self.push_subscribers.lock().await;
            let expired = subscribers
                .iter()
                .filter_map(|(handle, subscriber)| {
                    now.checked_duration_since(subscriber.registered_at)
                        .is_some_and(|age| age >= PUSH_SUBSCRIBER_MAX_AGE)
                        .then_some(handle.clone())
                })
                .collect::<Vec<_>>();
            expired
                .into_iter()
                .filter_map(|handle| subscribers.remove(&handle))
                .collect::<Vec<_>>()
        };
        let count = removed.len();
        for subscriber in removed {
            subscriber.close_modern_resource();
        }
        count
    }

    /// 当前活跃订阅数。生产诊断使用 `push_subscribers_summary`，这里只作为
    /// state 单测的轻量断言入口保留。
    #[cfg(test)]
    pub async fn push_subscriber_count(&self) -> usize {
        self.push_subscribers.lock().await.len()
    }

    #[cfg(test)]
    pub async fn enqueue_modern_push_for_test(
        &self,
        handle: &str,
        event: serde_json::Value,
    ) -> Result<(), String> {
        let resource = self
            .push_subscribers
            .lock()
            .await
            .get(handle)
            .and_then(|subscriber| match &subscriber.delivery {
                SubscriberDelivery::ModernResource(resource) => Some(resource.clone()),
                SubscriberDelivery::LegacyPeer(_, _) => None,
            })
            .ok_or_else(|| "modern push resource missing".to_string())?;
        if let Some(work) = resource.enqueue(event) {
            resource.send_pending_notification(work).await;
        }
        Ok(())
    }

    #[cfg(test)]
    pub async fn dispatch_push_for_test(
        &self,
        handle: &str,
        proto_id: u32,
        body: &[u8],
    ) -> Result<bool, String> {
        let subscriber = self
            .push_subscribers
            .lock()
            .await
            .get(handle)
            .cloned()
            .ok_or_else(|| "push subscriber missing".to_string())?;
        let decode_result = super::classify_trade_push(proto_id, body);
        let (push_acc_id, push_market, decode_status, event_type) = match &decode_result {
            super::TradePushDecode::NotTrade => (None, None, "ok", "quote"),
            super::TradePushDecode::Decoded { acc_id, trd_market } => {
                (Some(*acc_id), Some(*trd_market), "ok", "trade")
            }
            super::TradePushDecode::DecodeFailed => (None, None, "failed", "trade"),
        };
        let push_market_str = push_market.map(super::trd_market_int_to_str);
        let registry = futu_auth_pipeline::FilterRegistry::with_defaults();
        if !super::push_delivery_is_authorized(
            &subscriber,
            &self.key_store,
            &registry,
            &decode_result,
            event_type,
            push_acc_id,
            push_market_str,
            proto_id,
        ) {
            return Ok(false);
        }
        let payload = serde_json::json!({
            "kind": "futu_push",
            "proto_id": proto_id,
            "acc_id": push_acc_id,
            "event_type": event_type,
            "trd_market": push_market_str,
            "decode_status": decode_status,
            "body_base64": super::base64_encode_bytes(body),
        });
        match subscriber.delivery {
            SubscriberDelivery::LegacyPeer(peer, minimum_level) => {
                super::notify_legacy_push(&peer, &minimum_level, payload)
                    .await
                    .map_err(|error| error.to_string())?;
            }
            SubscriberDelivery::ModernResource(resource) => {
                if let Some(work) = resource.enqueue(payload) {
                    resource.send_pending_notification(work).await;
                }
            }
        }
        Ok(true)
    }

    #[cfg(test)]
    pub async fn expire_push_subscriber_for_test(&self, handle: &str) {
        if let Some(subscriber) = self.push_subscribers.lock().await.get_mut(handle)
            && let Some(expired_at) = Instant::now()
                .checked_sub(PUSH_SUBSCRIBER_MAX_AGE + std::time::Duration::from_secs(1))
        {
            subscriber.registered_at = expired_at;
        }
    }

    #[cfg(test)]
    pub async fn set_push_subscriber_registered_at_for_test(
        &self,
        handle: &str,
        registered_at: Instant,
    ) {
        if let Some(subscriber) = self.push_subscribers.lock().await.get_mut(handle) {
            subscriber.registered_at = registered_at;
        }
    }

    #[cfg(test)]
    pub async fn purge_expired_push_subscribers_at_for_test(&self, now: Instant) -> usize {
        self.purge_expired_push_subscribers_at(now).await
    }

    #[cfg(test)]
    pub async fn push_subscriber_exists_for_test(&self, handle: &str) -> bool {
        self.push_subscribers.lock().await.contains_key(handle)
    }

    #[cfg(test)]
    pub async fn modern_queue_stats_for_test(&self, handle: &str) -> Option<(usize, u64)> {
        let resource =
            self.push_subscribers.lock().await.get(handle).and_then(
                |subscriber| match &subscriber.delivery {
                    SubscriberDelivery::ModernResource(resource) => Some(resource.clone()),
                    SubscriberDelivery::LegacyPeer(_, _) => None,
                },
            )?;
        let inner = resource.lock_inner();
        Some((inner.queue.len(), inner.dropped_events))
    }

    #[cfg(test)]
    pub async fn modern_notification_stats_for_test(&self, handle: &str) -> Option<(u64, u64)> {
        let resource =
            self.push_subscribers.lock().await.get(handle).and_then(
                |subscriber| match &subscriber.delivery {
                    SubscriberDelivery::ModernResource(resource) => Some(resource.clone()),
                    SubscriberDelivery::LegacyPeer(_, _) => None,
                },
            )?;
        let inner = resource.lock_inner();
        Some((
            inner.notification_send_attempts,
            inner.notification_send_failures,
        ))
    }

    #[cfg(test)]
    pub async fn poison_modern_queue_for_test(&self, handle: &str) -> Result<(), String> {
        let resource = self
            .push_subscribers
            .lock()
            .await
            .get(handle)
            .and_then(|subscriber| match &subscriber.delivery {
                SubscriberDelivery::ModernResource(resource) => Some(Arc::clone(resource)),
                SubscriberDelivery::LegacyPeer(_, _) => None,
            })
            .ok_or_else(|| "modern push resource missing".to_string())?;
        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _guard = match resource.inner.lock() {
                Ok(inner) => inner,
                Err(poisoned) => poisoned.into_inner(),
            };
            std::panic::resume_unwind(Box::new("intentional modern queue poison"));
        }));
        if outcome.is_ok() || !resource.inner.is_poisoned() {
            return Err("failed to poison modern push queue".to_string());
        }
        Ok(())
    }

    #[cfg(test)]
    pub async fn modern_queue_is_poisoned_for_test(&self, handle: &str) -> Option<bool> {
        self.push_subscribers
            .lock()
            .await
            .get(handle)
            .and_then(|subscriber| match &subscriber.delivery {
                SubscriberDelivery::ModernResource(resource) => Some(resource.inner.is_poisoned()),
                SubscriberDelivery::LegacyPeer(_, _) => None,
            })
    }

    #[cfg(test)]
    pub async fn pause_next_modern_notification_after_send_for_test(
        &self,
        handle: &str,
        sent: Arc<tokio::sync::Barrier>,
        resume: Arc<tokio::sync::Barrier>,
    ) -> Result<(), String> {
        let resource = self
            .push_subscribers
            .lock()
            .await
            .get(handle)
            .and_then(|subscriber| match &subscriber.delivery {
                SubscriberDelivery::ModernResource(resource) => Some(Arc::clone(resource)),
                SubscriberDelivery::LegacyPeer(_, _) => None,
            })
            .ok_or_else(|| "modern push resource missing".to_string())?;
        resource.lock_inner().notification_send_pause = Some((sent, resume));
        Ok(())
    }

    #[cfg(test)]
    pub async fn close_modern_resource_without_removing_for_test(
        &self,
        handle: &str,
    ) -> Result<(), String> {
        let resource = self
            .push_subscribers
            .lock()
            .await
            .get(handle)
            .and_then(|subscriber| match &subscriber.delivery {
                SubscriberDelivery::ModernResource(resource) => Some(Arc::clone(resource)),
                SubscriberDelivery::LegacyPeer(_, _) => None,
            })
            .ok_or_else(|| "modern push resource missing".to_string())?;
        resource.close();
        Ok(())
    }

    /// v1.4.58 Phase A: 列出所有 push 订阅 summary（tool diagnostic 用）。
    ///
    /// 返 Vec<(session_id, acc_ids, age_secs)>。
    ///
    /// **MED-NEW-3 修（2nd review）**：加 `caller_allowed_acc_ids` 参数做
    /// scope-mode 多租过滤。当 caller 的 key 有 `allowed_acc_ids` 白名单时，
    /// **只返** subscription 的 `acc_ids` 与 caller 白名单有交集的条目。
    /// 避免 agent A（acc_ids=[100, 200]）通过本 tool 看到 agent B 订阅的
    /// acc_id=[300, 400]。
    ///
    /// `caller_allowed_acc_ids=None` / empty → 不过滤（legacy mode / no-scope key）。
    ///
    /// **rmcp 版本兼容**：rmcp 1.4.0 `Peer<RoleServer>` 不实装 `PartialEq`，
    /// 无法按 peer 身份直接过滤。若未来 rmcp 加 PartialEq，可切到更精确的
    /// per-session-owner 过滤（当前只能靠 acc_id 权限交集近似）。
    pub async fn push_subscribers_summary(
        &self,
        caller_allowed_acc_ids: Option<&HashSet<u64>>,
    ) -> Vec<(String, HashSet<u64>, u64)> {
        let subs = self.push_subscribers.lock().await;
        let now = Instant::now();
        subs.iter()
            .filter(|(_, sub)| subscriber_visible_to_caller(&sub.acc_ids, caller_allowed_acc_ids))
            .map(|(id, sub)| {
                let age = now
                    .checked_duration_since(sub.registered_at)
                    .map(|d| d.as_secs())
                    .unwrap_or(0);
                // LOW-3RD-1（3rd code review）：scope mode 下返回的 acc_ids 要与
                // caller allowed 求交集 — 防止 caller=[100] 看到 sub=[100, 999]
                // 时知道 999 这个 acc 存在。sub.acc_ids 空集（subscribe-all）不做
                // 交集（概念上 caller 看到的是"有个 catch-all subscriber"，不泄漏
                // 具体账户信息）。
                let visible_accs = match caller_allowed_acc_ids {
                    Some(allowed) if !allowed.is_empty() && !sub.acc_ids.is_empty() => {
                        sub.acc_ids.intersection(allowed).copied().collect()
                    }
                    _ => sub.acc_ids.clone(),
                };
                (id.clone(), visible_accs, age)
            })
            .collect()
    }
}

impl PushSubscriber {
    pub(super) fn close_modern_resource(&self) {
        if let SubscriberDelivery::ModernResource(resource) = &self.delivery {
            resource.close();
        }
    }
}