功能(提醒)：统一项目提醒中心与桌面通知链路

增加通用提醒状态、数据库迁移和定时同步，覆盖风险时效、文件回执、项目里程碑、访视窗口与协作申请。

新增网页端和桌面端提醒中心、真实投递诊断与固定隐私通知正文，并补齐登录来源聚合、测试和说明文档。

backend/app/services/collaboration_service.py

@@ -1139,6 +1139,7 @@ async def create_edit_request(
         source_type="COLLABORATION_EDIT_REQUEST",
         source_id=str(request.id),
         dedupe_key=f"collaboration-edit-request:{request.id}",
+        source_version="PENDING",
     )
     await db.commit()
     await db.refresh(request)

backend/app/services/desktop_notification_service.py

@@ -10,12 +10,9 @@ from app.models.desktop_notification import (
     DesktopNotificationDelivery,
     DesktopNotificationSubscription,
 )
-from app.models.distribution import Distribution, DistributionStatus, DistributionTargetType
-from app.models.document import Document
-from app.models.document_version import DocumentVersion
-from app.models.study import Study
+from app.models.notification import Notification
 from app.models.study_member import StudyMember
-from app.schemas.notification import NotificationItem
+from app.models.user import User
 
 CLAIM_LEASE = timedelta(minutes=5)
 
@@ -47,46 +44,28 @@ async def set_subscription(
 
 
 def _eligible_query(user_id: uuid.UUID, enabled_at: datetime, lease_cutoff: datetime):
-    role_target_exists = exists(
+    active_membership = exists(
         select(StudyMember.id).where(
-            StudyMember.study_id == Document.trial_id,
+            StudyMember.study_id == Notification.study_id,
             StudyMember.user_id == user_id,
             StudyMember.is_active.is_(True),
-            StudyMember.role_in_study == Distribution.target_id,
         )
     )
-    target_matches = or_(
-        and_(
-            Distribution.target_type == DistributionTargetType.USER,
-            Distribution.target_id == str(user_id),
-        ),
-        and_(
-            Distribution.target_type == DistributionTargetType.ROLE,
-            role_target_exists,
-        ),
-    )
     return (
-        select(
-            Distribution,
-            Document,
-            DocumentVersion,
-            Study,
-            DesktopNotificationDelivery,
-        )
-        .join(Document, Distribution.document_id == Document.id)
-        .join(DocumentVersion, Distribution.version_id == DocumentVersion.id)
-        .join(Study, Document.trial_id == Study.id)
+        select(Notification, DesktopNotificationDelivery)
         .outerjoin(
             DesktopNotificationDelivery,
             and_(
-                DesktopNotificationDelivery.distribution_id == Distribution.id,
+                DesktopNotificationDelivery.notification_id == Notification.id,
                 DesktopNotificationDelivery.user_id == user_id,
             ),
         )
         .where(
-            Distribution.status == DistributionStatus.ACTIVE,
-            Distribution.created_at >= enabled_at,
-            target_matches,
+            Notification.recipient_id == user_id,
+            Notification.created_at >= enabled_at,
+            Notification.resolved_at.is_(None),
+            Notification.read_at.is_(None),
+            active_membership,
             or_(
                 DesktopNotificationDelivery.id.is_(None),
                 and_(
@@ -98,8 +77,8 @@ def _eligible_query(user_id: uuid.UUID, enabled_at: datetime, lease_cutoff: date
                 ),
             ),
         )
-        .order_by(Distribution.created_at.asc())
-        .with_for_update(of=Distribution, skip_locked=True)
+        .order_by(Notification.created_at.asc())
+        .with_for_update(of=Notification, skip_locked=True)
     )
 
 
@@ -107,45 +86,46 @@ async def claim_notifications(
     db: AsyncSession,
     user_id: uuid.UUID,
     limit: int,
-) -> tuple[uuid.UUID | None, datetime | None, list[NotificationItem]]:
+) -> tuple[uuid.UUID | None, datetime | None, list[Notification]]:
     subscription = await get_subscription(db, user_id)
     if not subscription or not subscription.enabled or not subscription.enabled_at:
         return None, None, []
 
+    # Reconcile every active project before selecting desktop deliveries so permission
+    # changes fail closed and time-based reminders do not depend on opening the web Feed.
+    user = await db.get(User, user_id)
+    if user is None or not user.is_active:
+        return None, None, []
+    study_ids = (await db.scalars(
+        select(StudyMember.study_id).where(
+            StudyMember.user_id == user_id,
+            StudyMember.is_active.is_(True),
+        )
+    )).all()
+    from app.services.project_reminder_service import sync_project_reminders
+
+    for study_id in set(study_ids):
+        await sync_project_reminders(db, study_id, user)
+
     now = datetime.now(timezone.utc)
     token = uuid.uuid4()
     rows = (
         await db.execute(
-            _eligible_query(user_id, subscription.enabled_at, now - CLAIM_LEASE).limit(max(1, min(limit, 50)))
+            _eligible_query(user_id, subscription.enabled_at, now - CLAIM_LEASE)
+            .limit(max(1, min(limit, 50)))
         )
     ).all()
-    items: list[NotificationItem] = []
-    for distribution, document, version, study, delivery in rows:
+    items: list[Notification] = []
+    for notification, delivery in rows:
         if delivery is None:
             delivery = DesktopNotificationDelivery(
                 user_id=user_id,
-                distribution_id=distribution.id,
+                notification_id=notification.id,
             )
             db.add(delivery)
         delivery.claim_token = token
         delivery.claimed_at = now
-        items.append(
-            NotificationItem(
-                id=distribution.id,
-                document_id=document.id,
-                version_id=version.id,
-                document_title=document.title,
-                document_no=document.doc_no,
-                version_no=version.version_no,
-                change_summary=version.change_summary,
-                effective_at=version.effective_at,
-                created_at=distribution.created_at,
-                study_id=study.id,
-                study_name=study.name,
-                delivered_at=delivery.delivered_at,
-                read_at=delivery.read_at,
-            )
-        )
+        items.append(notification)
     await db.commit()
     if not items:
         return None, None, []
@@ -164,7 +144,7 @@ async def acknowledge_notifications(
             .where(
                 DesktopNotificationDelivery.user_id == user_id,
                 DesktopNotificationDelivery.claim_token == claim_token,
-                DesktopNotificationDelivery.distribution_id.in_(delivered_ids),
+                DesktopNotificationDelivery.notification_id.in_(delivered_ids),
             )
             .values(delivered_at=datetime.now(timezone.utc))
         )
@@ -174,21 +154,17 @@ async def acknowledge_notifications(
 async def mark_notification_read(
     db: AsyncSession,
     user_id: uuid.UUID,
-    distribution_id: uuid.UUID,
+    notification_id: uuid.UUID,
 ) -> None:
-    delivery = (
-        await db.execute(
-            select(DesktopNotificationDelivery).where(
-                DesktopNotificationDelivery.user_id == user_id,
-                DesktopNotificationDelivery.distribution_id == distribution_id,
-            )
-        )
-    ).scalar_one_or_none()
-    if delivery is None:
-        delivery = DesktopNotificationDelivery(
-            user_id=user_id,
-            distribution_id=distribution_id,
-        )
-        db.add(delivery)
-    delivery.read_at = datetime.now(timezone.utc)
+    item = await db.scalar(select(Notification).where(
+        Notification.id == notification_id,
+        Notification.recipient_id == user_id,
+        Notification.resolved_at.is_(None),
+    ))
+    if item is None:
+        return
+    now = datetime.now(timezone.utc)
+    item.read_at = item.read_at or now
+    if not item.requires_action:
+        item.resolved_at = item.resolved_at or now
     await db.commit()

backend/app/services/document_service.py

@@ -11,7 +11,7 @@ from urllib.parse import quote
 import aiofiles
 from fastapi import HTTPException, UploadFile, status
 from fastapi.responses import FileResponse
-from sqlalchemy import and_, delete as sa_delete, or_, select, update as sa_update
+from sqlalchemy import String, and_, cast, delete as sa_delete, or_, select, update as sa_update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.deps import get_cra_site_scope
@@ -32,6 +32,7 @@ from app.models.distribution import Distribution, DistributionStatus, Distributi
 from app.models.document import Document, DocumentScopeType, DocumentStatus
 from app.models.document_version import DocumentVersion, DocumentVersionStatus
 from app.models.desktop_notification import DesktopNotificationDelivery
+from app.models.notification import Notification
 from app.models.study import Study
 from app.schemas.acknowledgement import AcknowledgementCreate
 from app.schemas.distribution import DistributionCreate, DistributionRead, DistributionStats
@@ -39,6 +40,7 @@ from app.schemas.document import DocumentCreate, DocumentDetail, DocumentSummary
 from app.schemas.document_version import DocumentVersionRead, DocumentVersionSummary
 from app.schemas.notification import NotificationItem
 from app.schemas.user import UserDisplay
+from app.services import notification_service
 
 UPLOAD_ROOT = Path(__file__).resolve().parent.parent / "uploads" / "documents"
 
@@ -688,6 +690,10 @@ async def create_acknowledgement(
     if distribution.target_type == DistributionTargetType.ROLE:
         if distribution.target_id not in ("ADMIN" if is_system_admin(current_user) else "", getattr(membership, "role_in_study", "")):
             raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不在分发范围内")
+    if distribution.target_type == DistributionTargetType.SITE and not is_system_admin(current_user):
+        site_ids = await site_crud.list_ids_by_contact_user(db, doc.trial_id, current_user.id)
+        if distribution.target_id not in {str(site_id) for site_id in site_ids}:
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不在分发中心范围内")
 
     if payload.ack_type != AcknowledgementType.RECEIVED:
         raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="仅支持已接收回执")
@@ -722,6 +728,12 @@ async def create_acknowledgement(
             operator_role=await get_operator_role_label(db, doc.trial_id, current_user),
         )
     )
+    await notification_service.resolve_source_notifications(
+        db,
+        source_type="DOCUMENT_DISTRIBUTION",
+        source_id=str(distribution.id),
+        recipient_id=current_user.id,
+    )
     await db.commit()
     await db.refresh(ack)
     return ack
@@ -812,10 +824,18 @@ async def list_distribution_notifications(
         .join(DocumentVersion, Distribution.version_id == DocumentVersion.id)
         .join(Document, Distribution.document_id == Document.id)
         .join(Study, Document.trial_id == Study.id)
+        .outerjoin(
+            Notification,
+            and_(
+                Notification.source_type == "DOCUMENT_DISTRIBUTION",
+                Notification.source_id == cast(Distribution.id, String),
+                Notification.recipient_id == current_user.id,
+            ),
+        )
         .outerjoin(
             DesktopNotificationDelivery,
             and_(
-                DesktopNotificationDelivery.distribution_id == Distribution.id,
+                DesktopNotificationDelivery.notification_id == Notification.id,
                 DesktopNotificationDelivery.user_id == current_user.id,
             ),
         )

backend/app/services/notification_scheduler.py

@@ -0,0 +1,78 @@
+"""Server-side materialization for time and state driven business reminders."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from sqlalchemy import select, text
+
+from app.core.config import settings
+from app.db.session import SessionLocal
+from app.models.study_member import StudyMember
+from app.models.user import User, UserStatus
+from app.services.project_reminder_service import sync_project_reminders
+
+logger = logging.getLogger("ctms.notifications")
+NOTIFICATION_SYNC_LOCK_KEY = 0x43544D53  # "CTMS"
+
+
+async def sync_all_project_reminders_once() -> int:
+    """Synchronize every active project member once, with a cross-worker advisory lock."""
+    async with SessionLocal() as lock_session:
+        acquired = bool(await lock_session.scalar(
+            text("SELECT pg_try_advisory_lock(:lock_key)"),
+            {"lock_key": NOTIFICATION_SYNC_LOCK_KEY},
+        ))
+        if not acquired:
+            return 0
+        try:
+            memberships = (await lock_session.execute(
+                select(StudyMember.study_id, StudyMember.user_id)
+                .join(User, User.id == StudyMember.user_id)
+                .where(
+                    StudyMember.is_active.is_(True),
+                    User.status == UserStatus.ACTIVE,
+                )
+            )).all()
+            await lock_session.commit()
+            synced = 0
+            for study_id, user_id in memberships:
+                async with SessionLocal() as session:
+                    try:
+                        user = await session.get(User, user_id)
+                        if user is None or not user.is_active:
+                            continue
+                        await sync_project_reminders(session, study_id, user)
+                        synced += 1
+                    except Exception:
+                        await session.rollback()
+                        logger.exception(
+                            "Failed to synchronize reminders for study=%s user=%s",
+                            study_id,
+                            user_id,
+                        )
+            return synced
+        finally:
+            await lock_session.execute(
+                text("SELECT pg_advisory_unlock(:lock_key)"),
+                {"lock_key": NOTIFICATION_SYNC_LOCK_KEY},
+            )
+
+
+async def run_notification_sync_job(stop_event: asyncio.Event) -> None:
+    logger.info("Notification synchronization task started")
+    while not stop_event.is_set():
+        try:
+            synced = await sync_all_project_reminders_once()
+            logger.debug("Synchronized reminders for %s project memberships", synced)
+        except Exception:
+            logger.exception("Notification synchronization task failed")
+        try:
+            await asyncio.wait_for(
+                stop_event.wait(),
+                timeout=settings.NOTIFICATION_SYNC_INTERVAL_SECONDS,
+            )
+        except asyncio.TimeoutError:
+            continue
+    logger.info("Notification synchronization task stopped")

backend/app/services/notification_service.py

@@ -8,10 +8,24 @@ from fastapi import HTTPException, status
 from sqlalchemy import func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.models.desktop_notification import DesktopNotificationDelivery
 from app.models.notification import Notification
 from app.schemas.notification import GeneralNotificationFeed
 
 
+async def _requeue_desktop_delivery(db: AsyncSession, notification_id: uuid.UUID) -> None:
+    """Allow an escalated/reopened reminder to be delivered again by the desktop channel."""
+    await db.execute(
+        update(DesktopNotificationDelivery)
+        .where(DesktopNotificationDelivery.notification_id == notification_id)
+        .values(
+            claim_token=None,
+            claimed_at=None,
+            delivered_at=None,
+        )
+    )
+
+
 async def create_recipient_notifications(
     db: AsyncSession,
     *,
@@ -25,6 +39,9 @@ async def create_recipient_notifications(
     source_type: str,
     source_id: str,
     dedupe_key: str,
+    source_version: str | None = None,
+    requires_action: bool = True,
+    due_at: datetime | None = None,
 ) -> None:
     recipients = set(recipient_ids)
     if not recipients:
@@ -46,10 +63,78 @@ async def create_recipient_notifications(
             action_path=action_path,
             source_type=source_type,
             source_id=source_id,
+            source_version=source_version,
             dedupe_key=dedupe_key,
+            requires_action=requires_action,
+            due_at=due_at,
         ))
 
 
+async def sync_state_notification(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    recipient_id: uuid.UUID,
+    category: str,
+    priority: str,
+    title: str,
+    message: str,
+    action_path: str,
+    source_type: str,
+    source_id: str,
+    source_version: str,
+    active: bool,
+    due_at: datetime | None = None,
+    requires_action: bool = True,
+    dedupe_key: str | None = None,
+) -> None:
+    """Synchronize one actionable business state into a recipient reminder."""
+    dedupe_key = dedupe_key or f"state:{source_type}:{source_id}"
+    item = await db.scalar(select(Notification).where(
+        Notification.recipient_id == recipient_id,
+        Notification.dedupe_key == dedupe_key,
+    ))
+    now = datetime.now(timezone.utc)
+    if not active:
+        if item and item.resolved_at is None:
+            item.resolved_at = now
+            item.read_at = item.read_at or now
+        return
+    if item is None:
+        db.add(Notification(
+            study_id=study_id,
+            recipient_id=recipient_id,
+            category=category,
+            priority=priority,
+            title=title,
+            message=message,
+            action_path=action_path,
+            source_type=source_type,
+            source_id=source_id,
+            source_version=source_version,
+            dedupe_key=dedupe_key,
+            requires_action=requires_action,
+            due_at=due_at,
+        ))
+        return
+
+    should_notify_again = item.resolved_at is not None or item.source_version != source_version
+    item.study_id = study_id
+    item.category = category
+    item.priority = priority
+    item.title = title
+    item.message = message
+    item.action_path = action_path
+    item.source_version = source_version
+    item.requires_action = requires_action
+    item.due_at = due_at
+    if should_notify_again:
+        item.read_at = None
+        item.created_at = now
+        await _requeue_desktop_delivery(db, item.id)
+    item.resolved_at = None
+
+
 async def sync_aggregate_notification(
     db: AsyncSession,
     *,
@@ -63,6 +148,7 @@ async def sync_aggregate_notification(
     source_type: str,
     source_id: str,
     count: int,
+    due_at: datetime | None = None,
 ) -> None:
     dedupe_key = f"aggregate:{source_type}:{source_id}"
     item = await db.scalar(select(Notification).where(
@@ -89,32 +175,83 @@ async def sync_aggregate_notification(
             source_id=source_id,
             source_version=version,
             dedupe_key=dedupe_key,
+            requires_action=True,
+            due_at=due_at,
         ))
         return
     previous_count = int(item.source_version or 0)
+    item.category = category
     item.priority = priority
     item.title = title
     item.message = message
     item.action_path = action_path
     item.source_version = version
+    item.due_at = due_at
     if item.resolved_at is not None or count > previous_count:
         item.read_at = None
         item.created_at = now
+        await _requeue_desktop_delivery(db, item.id)
     item.resolved_at = None
 
 
+async def resolve_missing_recipient_sources(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    recipient_id: uuid.UUID,
+    source_type: str,
+    active_source_ids: set[str],
+) -> None:
+    filters = [
+        Notification.study_id == study_id,
+        Notification.recipient_id == recipient_id,
+        Notification.source_type == source_type,
+        Notification.resolved_at.is_(None),
+    ]
+    if active_source_ids:
+        filters.append(Notification.source_id.not_in(active_source_ids))
+    now = datetime.now(timezone.utc)
+    await db.execute(
+        update(Notification)
+        .where(*filters)
+        .values(resolved_at=now, read_at=func.coalesce(Notification.read_at, now))
+    )
+
+
 async def resolve_source_notifications(
     db: AsyncSession,
     *,
     source_type: str,
     source_id: str,
+    recipient_id: uuid.UUID | None = None,
+) -> None:
+    filters = [
+        Notification.source_type == source_type,
+        Notification.source_id == source_id,
+        Notification.resolved_at.is_(None),
+    ]
+    if recipient_id is not None:
+        filters.append(Notification.recipient_id == recipient_id)
+    now = datetime.now(timezone.utc)
+    await db.execute(
+        update(Notification)
+        .where(*filters)
+        .values(resolved_at=now, read_at=func.coalesce(Notification.read_at, now))
+    )
+
+
+async def resolve_recipient_study_notifications(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    recipient_id: uuid.UUID,
 ) -> None:
     now = datetime.now(timezone.utc)
     await db.execute(
         update(Notification)
         .where(
-            Notification.source_type == source_type,
-            Notification.source_id == source_id,
+            Notification.study_id == study_id,
+            Notification.recipient_id == recipient_id,
             Notification.resolved_at.is_(None),
         )
         .values(resolved_at=now, read_at=func.coalesce(Notification.read_at, now))
@@ -126,7 +263,11 @@ async def list_feed(
     *,
     study_id: uuid.UUID,
     recipient_id: uuid.UUID,
+    skip: int = 0,
     limit: int = 10,
+    category: str | None = None,
+    unread_only: bool = False,
+    requires_action: bool | None = None,
 ) -> GeneralNotificationFeed:
     active_filter = (
         Notification.study_id == study_id,
@@ -136,13 +277,36 @@ async def list_feed(
     unread_count = int(await db.scalar(
         select(func.count(Notification.id)).where(*active_filter, Notification.read_at.is_(None))
     ) or 0)
+
+    filters = [
+        Notification.study_id == study_id,
+        Notification.recipient_id == recipient_id,
+        Notification.resolved_at.is_(None),
+    ]
+    if category:
+        filters.append(Notification.category == category)
+    if unread_only:
+        filters.append(Notification.read_at.is_(None))
+    if requires_action is not None:
+        filters.append(Notification.requires_action.is_(requires_action))
+    total_count = int(await db.scalar(
+        select(func.count(Notification.id)).where(*filters)
+    ) or 0)
     items = (await db.scalars(
         select(Notification)
-        .where(*active_filter)
-        .order_by(Notification.read_at.is_not(None), Notification.created_at.desc())
-        .limit(max(1, min(limit, 50)))
+        .where(*filters)
+        .order_by(
+            Notification.read_at.is_not(None),
+            Notification.created_at.desc(),
+        )
+        .offset(max(0, skip))
+        .limit(max(1, min(limit, 100)))
     )).all()
-    return GeneralNotificationFeed(unread_count=unread_count, items=list(items))
+    return GeneralNotificationFeed(
+        unread_count=unread_count,
+        total_count=total_count,
+        items=list(items),
+    )
 
 
 async def mark_read(
@@ -161,7 +325,10 @@ async def mark_read(
     if item is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="通知不存在")
     if item.read_at is None:
-        item.read_at = datetime.now(timezone.utc)
+        now = datetime.now(timezone.utc)
+        item.read_at = now
+        if not item.requires_action:
+            item.resolved_at = now
         await db.commit()
         await db.refresh(item)
     return item
@@ -173,6 +340,7 @@ async def mark_all_read(
     study_id: uuid.UUID,
     recipient_id: uuid.UUID,
 ) -> None:
+    now = datetime.now(timezone.utc)
     await db.execute(
         update(Notification)
         .where(
@@ -180,7 +348,19 @@ async def mark_all_read(
             Notification.recipient_id == recipient_id,
             Notification.resolved_at.is_(None),
             Notification.read_at.is_(None),
+            Notification.requires_action.is_(False),
         )
-        .values(read_at=datetime.now(timezone.utc))
+        .values(read_at=now, resolved_at=now)
+    )
+    await db.execute(
+        update(Notification)
+        .where(
+            Notification.study_id == study_id,
+            Notification.recipient_id == recipient_id,
+            Notification.resolved_at.is_(None),
+            Notification.read_at.is_(None),
+            Notification.requires_action.is_(True),
+        )
+        .values(read_at=now)
     )
     await db.commit()

backend/app/services/project_reminder_service.py

@@ -1,61 +1,152 @@
 from __future__ import annotations
 
 import uuid
+from datetime import date, datetime, time, timedelta, timezone
+from zoneinfo import ZoneInfo
 
+from sqlalchemy import and_, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.deps import get_cra_site_scope, is_system_admin
 from app.core.project_permissions import role_has_api_permission
-from app.crud import ae as ae_crud
 from app.crud import member as member_crud
-from app.crud import monitoring_visit_issue as monitoring_issue_crud
+from app.crud import site as site_crud
+from app.models.acknowledgement import Acknowledgement, AcknowledgementType
+from app.models.ae import AdverseEvent
+from app.models.collaboration import CollaborationEditRequest, CollaborationFile, CollaborationMember
+from app.models.distribution import Distribution, DistributionStatus, DistributionTargetType
+from app.models.document import Document, DocumentStatus
+from app.models.document_version import DocumentVersion
+from app.models.milestone import Milestone
+from app.models.monitoring_visit_issue import MonitoringVisitIssue
+from app.models.site import Site
+from app.models.subject import Subject
+from app.models.user import User
+from app.models.visit import Visit
 from app.services import notification_service
 
+BUSINESS_TIMEZONE = ZoneInfo("Asia/Shanghai")
+RISK_DUE_SOON_DAYS = 3
+MILESTONE_DUE_SOON_DAYS = 7
+VISIT_WINDOW_DUE_SOON_DAYS = 3
 
-async def sync_project_reminders(db: AsyncSession, study_id: uuid.UUID, user) -> None:
-    membership = await member_crud.get_member(db, study_id, user.id)
-    if not is_system_admin(user) and (not membership or not membership.is_active):
-        return
-    role = membership.role_in_study if membership else ""
+
+def _date_due_at(value: date | None) -> datetime | None:
+    if value is None:
+        return None
+    return datetime.combine(value, time.max, tzinfo=BUSINESS_TIMEZONE).astimezone(timezone.utc)
+
+
+def _as_utc(value: datetime | None) -> datetime | None:
+    if value is None:
+        return None
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+async def _count_and_earliest(db: AsyncSession, model, *filters):
+    return (await db.execute(
+        select(func.count(model.id), func.min(model.report_due_date if model is AdverseEvent else model.due_at))
+        .where(*filters)
+    )).one()
+
+
+async def _sync_risk_reminders(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    user,
+    role: str,
+    current_time: datetime,
+) -> None:
     can_read_aes = is_system_admin(user) or await role_has_api_permission(
         db, study_id, role, "subject_aes:read"
     )
     can_read_monitoring = is_system_admin(user) or await role_has_api_permission(
         db, study_id, role, "monitoring_issues:read"
     )
+    cra_scope = await get_cra_site_scope(db, study_id, user)
+    site_ids = cra_scope[0] if cra_scope else None
+    today = current_time.astimezone(BUSINESS_TIMEZONE).date()
+    due_soon_date = today + timedelta(days=RISK_DUE_SOON_DAYS)
 
+    ae_scope = [AdverseEvent.study_id == study_id, AdverseEvent.status != "CLOSED"]
+    if site_ids is not None:
+        ae_scope.append(AdverseEvent.site_id.in_(site_ids))
     overdue_aes = 0
+    overdue_ae_due = None
+    due_soon_aes = 0
+    due_soon_ae_due = None
     if can_read_aes:
-        cra_scope = await get_cra_site_scope(db, study_id, user)
-        items = await ae_crud.list_ae(
+        overdue_aes, overdue_ae_due = await _count_and_earliest(
             db,
-            study_id,
-            overdue=True,
-            site_ids=cra_scope[0] if cra_scope else None,
+            AdverseEvent,
+            *ae_scope,
+            AdverseEvent.report_due_date.is_not(None),
+            AdverseEvent.report_due_date < today,
+        )
+        due_soon_aes, due_soon_ae_due = await _count_and_earliest(
+            db,
+            AdverseEvent,
+            *ae_scope,
+            AdverseEvent.report_due_date >= today,
+            AdverseEvent.report_due_date <= due_soon_date,
         )
-        overdue_aes = len(items)
     await notification_service.sync_aggregate_notification(
         db,
         study_id=study_id,
         recipient_id=user.id,
         category="RISK_OVERDUE_AE",
-        priority="HIGH",
+        priority="URGENT",
         title="逾期 AE 待处理",
         message=f"当前有 {overdue_aes} 条逾期 AE 需要跟进",
         action_path="/risk-issues/sae",
         source_type="RISK_OVERDUE_AE",
         source_id=str(study_id),
-        count=overdue_aes if can_read_aes else 0,
+        count=int(overdue_aes) if can_read_aes else 0,
+        due_at=_date_due_at(overdue_ae_due),
+    )
+    await notification_service.sync_aggregate_notification(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        category="RISK_AE_DUE_SOON",
+        priority="HIGH",
+        title="AE 上报时限临近",
+        message=f"未来 {RISK_DUE_SOON_DAYS} 天内有 {due_soon_aes} 条 AE 到达上报时限",
+        action_path="/risk-issues/sae",
+        source_type="RISK_AE_DUE_SOON",
+        source_id=str(study_id),
+        count=int(due_soon_aes) if can_read_aes else 0,
+        due_at=_date_due_at(due_soon_ae_due),
     )
 
+    monitoring_scope = [
+        MonitoringVisitIssue.study_id == study_id,
+        MonitoringVisitIssue.status == "OPEN",
+        MonitoringVisitIssue.due_at.is_not(None),
+    ]
+    if site_ids is not None:
+        monitoring_scope.append(MonitoringVisitIssue.site_id.in_(site_ids))
     overdue_monitoring = 0
+    overdue_monitoring_due = None
+    due_soon_monitoring = 0
+    due_soon_monitoring_due = None
     if can_read_monitoring:
-        overdue_monitoring = len(await monitoring_issue_crud.list_issues(
+        overdue_monitoring, overdue_monitoring_due = await _count_and_earliest(
             db,
-            study_id,
-            overdue=True,
-            limit=2000,
-        ))
+            MonitoringVisitIssue,
+            *monitoring_scope,
+            MonitoringVisitIssue.due_at < current_time,
+        )
+        due_soon_monitoring, due_soon_monitoring_due = await _count_and_earliest(
+            db,
+            MonitoringVisitIssue,
+            *monitoring_scope,
+            MonitoringVisitIssue.due_at >= current_time,
+            MonitoringVisitIssue.due_at <= current_time + timedelta(days=RISK_DUE_SOON_DAYS),
+        )
     await notification_service.sync_aggregate_notification(
         db,
         study_id=study_id,
@@ -67,6 +158,384 @@ async def sync_project_reminders(db: AsyncSession, study_id: uuid.UUID, user) ->
         action_path="/risk-issues/monitoring-visits",
         source_type="RISK_OVERDUE_MONITORING",
         source_id=str(study_id),
-        count=overdue_monitoring if can_read_monitoring else 0,
+        count=int(overdue_monitoring) if can_read_monitoring else 0,
+        due_at=_as_utc(overdue_monitoring_due),
+    )
+    await notification_service.sync_aggregate_notification(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        category="RISK_MONITORING_DUE_SOON",
+        priority="HIGH",
+        title="监查问题整改时限临近",
+        message=f"未来 {RISK_DUE_SOON_DAYS} 天内有 {due_soon_monitoring} 条监查问题到期",
+        action_path="/risk-issues/monitoring-visits",
+        source_type="RISK_MONITORING_DUE_SOON",
+        source_id=str(study_id),
+        count=int(due_soon_monitoring) if can_read_monitoring else 0,
+        due_at=_as_utc(due_soon_monitoring_due),
+    )
+
+
+async def _sync_document_distribution_reminders(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    user,
+    role: str,
+    current_time: datetime,
+) -> None:
+    can_read_documents = is_system_admin(user) or await role_has_api_permission(
+        db, study_id, role, "documents:read"
+    )
+    if not can_read_documents:
+        await notification_service.resolve_missing_recipient_sources(
+            db,
+            study_id=study_id,
+            recipient_id=user.id,
+            source_type="DOCUMENT_DISTRIBUTION",
+            active_source_ids=set(),
+        )
+        return
+    site_ids = await site_crud.list_ids_by_contact_user(db, study_id, user.id)
+    target_filters = [
+        and_(
+            Distribution.target_type == DistributionTargetType.USER,
+            Distribution.target_id == str(user.id),
+        ),
+        and_(
+            Distribution.target_type == DistributionTargetType.ROLE,
+            Distribution.target_id == role,
+        ),
+    ]
+    if site_ids:
+        target_filters.append(and_(
+            Distribution.target_type == DistributionTargetType.SITE,
+            Distribution.target_id.in_({str(site_id) for site_id in site_ids}),
+        ))
+    rows = (await db.execute(
+        select(Distribution, Document, DocumentVersion)
+        .join(Document, Distribution.document_id == Document.id)
+        .join(DocumentVersion, Distribution.version_id == DocumentVersion.id)
+        .outerjoin(
+            Acknowledgement,
+            and_(
+                Acknowledgement.distribution_id == Distribution.id,
+                Acknowledgement.user_id == user.id,
+                Acknowledgement.ack_type == AcknowledgementType.RECEIVED,
+            ),
+        )
+        .where(
+            Document.trial_id == study_id,
+            Document.status == DocumentStatus.ACTIVE,
+            Distribution.status == DistributionStatus.ACTIVE,
+            Acknowledgement.id.is_(None),
+            or_(*target_filters),
+        )
+    )).all()
+
+    active_source_ids: set[str] = set()
+    for distribution, document, version in rows:
+        source_id = str(distribution.id)
+        active_source_ids.add(source_id)
+        distribution_due_at = _as_utc(distribution.due_at)
+        if distribution_due_at and distribution_due_at < current_time:
+            stage = "OVERDUE"
+            category = "DOCUMENT_ACK_OVERDUE"
+            priority = "URGENT"
+            title = "文件回执已逾期"
+        elif distribution_due_at and distribution_due_at <= current_time + timedelta(days=RISK_DUE_SOON_DAYS):
+            stage = "DUE_SOON"
+            category = "DOCUMENT_ACK_DUE_SOON"
+            priority = "HIGH"
+            title = "文件回执即将到期"
+        else:
+            stage = "PENDING"
+            category = "DOCUMENT_DISTRIBUTION"
+            priority = "NORMAL"
+            title = "有新的文件版本待接收"
+        due_version = distribution_due_at.isoformat() if distribution_due_at else "none"
+        await notification_service.sync_state_notification(
+            db,
+            study_id=study_id,
+            recipient_id=user.id,
+            category=category,
+            priority=priority,
+            title=title,
+            message=f"“{document.title}” {version.version_no} 已分发，请完成接收回执",
+            action_path=f"/documents/{document.id}",
+            source_type="DOCUMENT_DISTRIBUTION",
+            source_id=source_id,
+            source_version=f"{stage}:{due_version}",
+            active=True,
+            due_at=distribution_due_at,
+        )
+    await notification_service.resolve_missing_recipient_sources(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        source_type="DOCUMENT_DISTRIBUTION",
+        active_source_ids=active_source_ids,
+    )
+
+
+async def _sync_milestone_reminders(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    user,
+    role: str,
+    current_time: datetime,
+) -> None:
+    can_read = is_system_admin(user) or await role_has_api_permission(
+        db, study_id, role, "project_milestones:read"
+    )
+    active_source_ids: set[str] = set()
+    if can_read:
+        today = current_time.astimezone(BUSINESS_TIMEZONE).date()
+        due_soon_date = today + timedelta(days=MILESTONE_DUE_SOON_DAYS)
+        milestones = (await db.scalars(
+            select(Milestone).where(
+                Milestone.study_id == study_id,
+                Milestone.owner_id == user.id,
+                Milestone.status != "DONE",
+            )
+        )).all()
+        for milestone in milestones:
+            due_date = milestone.adjusted_end_date or milestone.planned_date
+            if due_date is None or due_date > due_soon_date:
+                continue
+            source_id = str(milestone.id)
+            active_source_ids.add(source_id)
+            overdue = due_date < today
+            await notification_service.sync_state_notification(
+                db,
+                study_id=study_id,
+                recipient_id=user.id,
+                category="MILESTONE_OVERDUE" if overdue else "MILESTONE_DUE_SOON",
+                priority="HIGH" if overdue else "NORMAL",
+                title="项目里程碑已逾期" if overdue else "项目里程碑即将到期",
+                message=f"“{milestone.name}”计划日期为 {due_date.isoformat()}",
+                action_path="/project/milestones",
+                source_type="PROJECT_MILESTONE",
+                source_id=source_id,
+                source_version=f"{'OVERDUE' if overdue else 'DUE_SOON'}:{due_date.isoformat()}:{milestone.status}",
+                active=True,
+                due_at=_date_due_at(due_date),
+            )
+    await notification_service.resolve_missing_recipient_sources(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        source_type="PROJECT_MILESTONE",
+        active_source_ids=active_source_ids,
+    )
+
+
+async def _sync_visit_window_reminders(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    user,
+    role: str,
+    current_time: datetime,
+) -> None:
+    """Project active visit windows to the members who can actually maintain them."""
+    can_manage_visits = is_system_admin(user) or await role_has_api_permission(
+        db, study_id, role, "visits:update"
+    )
+    active_source_ids: set[str] = set()
+    if can_manage_visits:
+        cra_scope = await get_cra_site_scope(db, study_id, user)
+        site_ids = cra_scope[0] if cra_scope else None
+        filters = [
+            Visit.study_id == study_id,
+            Visit.actual_date.is_(None),
+            Visit.status.not_in(["DONE", "CANCELLED"]),
+            Subject.status.not_in(["COMPLETED", "DROPPED"]),
+            Site.is_active.is_(True),
+        ]
+        if site_ids is not None:
+            filters.append(Subject.site_id.in_(site_ids))
+        rows = (await db.execute(
+            select(Visit, Subject)
+            .join(Subject, Subject.id == Visit.subject_id)
+            .join(Site, Site.id == Subject.site_id)
+            .where(*filters)
+        )).all()
+
+        today = current_time.astimezone(BUSINESS_TIMEZONE).date()
+        due_soon_date = today + timedelta(days=VISIT_WINDOW_DUE_SOON_DAYS)
+        for visit, subject in rows:
+            window_start = visit.window_start or visit.planned_date
+            window_end = visit.window_end or visit.planned_date
+            if window_start is None or window_end is None or window_start > window_end:
+                continue
+
+            if window_end < today:
+                category = "VISIT_WINDOW_MISSED"
+                priority = "HIGH"
+                title = "受试者访视已错过窗口"
+                message = "一个受试者访视已错过计划窗口，请进入详情核对并处理"
+                stage = "MISSED"
+            elif window_start <= due_soon_date:
+                category = "VISIT_WINDOW_DUE_SOON"
+                priority = "NORMAL"
+                title = "受试者访视窗口临近"
+                message = "一个受试者访视已进入近期窗口，请及时核对访视安排"
+                stage = "DUE_SOON"
+            else:
+                continue
+
+            source_id = str(visit.id)
+            active_source_ids.add(source_id)
+            await notification_service.sync_state_notification(
+                db,
+                study_id=study_id,
+                recipient_id=user.id,
+                category=category,
+                priority=priority,
+                title=title,
+                message=message,
+                action_path=f"/subjects/{subject.id}",
+                source_type="SUBJECT_VISIT_WINDOW",
+                source_id=source_id,
+                source_version=f"{stage}:{window_start.isoformat()}:{window_end.isoformat()}",
+                active=True,
+                due_at=_date_due_at(window_end),
+            )
+    await notification_service.resolve_missing_recipient_sources(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        source_type="SUBJECT_VISIT_WINDOW",
+        active_source_ids=active_source_ids,
+    )
+
+
+async def _sync_collaboration_edit_request_reminders(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    user,
+    role: str,
+) -> None:
+    can_read = is_system_admin(user) or await role_has_api_permission(
+        db, study_id, role, "collaboration:read"
+    )
+    if not can_read:
+        await notification_service.resolve_missing_recipient_sources(
+            db,
+            study_id=study_id,
+            recipient_id=user.id,
+            source_type="COLLABORATION_EDIT_REQUEST",
+            active_source_ids=set(),
+        )
+        return
+    manages_file = or_(
+        CollaborationFile.owner_id == user.id,
+        select(CollaborationMember.id).where(
+            CollaborationMember.file_id == CollaborationFile.id,
+            CollaborationMember.user_id == user.id,
+            CollaborationMember.role == "MANAGER",
+        ).exists(),
+    )
+    rows = (await db.execute(
+        select(CollaborationEditRequest, CollaborationFile, User)
+        .join(CollaborationFile, CollaborationFile.id == CollaborationEditRequest.file_id)
+        .join(User, User.id == CollaborationEditRequest.requester_id)
+        .where(
+            CollaborationFile.study_id == study_id,
+            CollaborationFile.status == "ACTIVE",
+            CollaborationFile.deleted_at.is_(None),
+            CollaborationEditRequest.status == "PENDING",
+            manages_file,
+        )
+    )).all()
+    active_source_ids: set[str] = set()
+    for request, item, requester in rows:
+        source_id = str(request.id)
+        active_source_ids.add(source_id)
+        requester_name = str(requester.full_name or requester.email or "项目成员")
+        await notification_service.sync_state_notification(
+            db,
+            study_id=study_id,
+            recipient_id=user.id,
+            category="COLLABORATION_EDIT_REQUEST",
+            priority="NORMAL",
+            title="新的编辑权限申请",
+            message=f"{requester_name} 申请编辑“{item.title}”",
+            action_path=f"/knowledge/collaboration?editRequestFile={item.id}",
+            source_type="COLLABORATION_EDIT_REQUEST",
+            source_id=source_id,
+            source_version="PENDING",
+            active=True,
+            dedupe_key=f"collaboration-edit-request:{request.id}",
+        )
+    await notification_service.resolve_missing_recipient_sources(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        source_type="COLLABORATION_EDIT_REQUEST",
+        active_source_ids=active_source_ids,
+    )
+
+
+async def sync_project_reminders(
+    db: AsyncSession,
+    study_id: uuid.UUID,
+    user,
+    *,
+    now: datetime | None = None,
+) -> None:
+    membership = await member_crud.get_member(db, study_id, user.id)
+    if not membership or not membership.is_active:
+        await notification_service.resolve_recipient_study_notifications(
+            db,
+            study_id=study_id,
+            recipient_id=user.id,
+        )
+        await db.commit()
+        return
+    role = membership.role_in_study
+    current_time = now or datetime.now(timezone.utc)
+    if current_time.tzinfo is None:
+        current_time = current_time.replace(tzinfo=timezone.utc)
+    else:
+        current_time = current_time.astimezone(timezone.utc)
+
+    await _sync_risk_reminders(
+        db,
+        study_id=study_id,
+        user=user,
+        role=role,
+        current_time=current_time,
+    )
+    await _sync_document_distribution_reminders(
+        db,
+        study_id=study_id,
+        user=user,
+        role=role,
+        current_time=current_time,
+    )
+    await _sync_milestone_reminders(
+        db,
+        study_id=study_id,
+        user=user,
+        role=role,
+        current_time=current_time,
+    )
+    await _sync_visit_window_reminders(
+        db,
+        study_id=study_id,
+        user=user,
+        role=role,
+        current_time=current_time,
+    )
+    await _sync_collaboration_edit_request_reminders(
+        db,
+        study_id=study_id,
+        user=user,
+        role=role,
     )
     await db.commit()

backend/app/services/user_login_sessions.py

@@ -195,7 +195,7 @@ async def get_login_summaries(
                 "last_login_at": row.login_at,
                 "last_seen_at": row.last_seen_at,
                 "client_type": row.client_type,
-                "active_session_count": 0,
+                "active_sources": set(),
             },
         )
         row_last_seen_at = _as_utc(row.last_seen_at)
@@ -204,14 +204,16 @@ async def get_login_summaries(
         ):
             current["last_seen_at"] = row_last_seen_at
         if row.ended_at is None and row_last_seen_at >= cutoff:
-            current["active_session_count"] += 1
+            source_key = (row.client_type, row.login_ip) if row.login_ip else ("session", row.id)
+            current["active_sources"].add(source_key)
     for user_id, item in mutable.items():
+        active_session_count = len(item["active_sources"])
         summaries[user_id] = UserLoginSummary(
-            status="ONLINE" if item["active_session_count"] else "OFFLINE",
+            status="ONLINE" if active_session_count else "OFFLINE",
             last_login_at=item["last_login_at"],
             last_seen_at=item["last_seen_at"],
             client_type=item["client_type"],
-            active_session_count=item["active_session_count"],
+            active_session_count=active_session_count,
         )
     return summaries