diff --git a/engine/apps/alerts/tasks/notify_user.py b/engine/apps/alerts/tasks/notify_user.py
index d9265571f2..10f62b257e 100644
--- a/engine/apps/alerts/tasks/notify_user.py
+++ b/engine/apps/alerts/tasks/notify_user.py
@@ -103,7 +103,7 @@ def notify_user_task(
     important=False,
     notify_anyway=False,
 ):
-    from apps.alerts.models import AlertGroup, UserHasNotification, UserNotificationBundle
+    from apps.alerts.models import AlertGroup, AlertGroupLogRecord, UserHasNotification, UserNotificationBundle
     from apps.base.models import UserNotificationPolicy, UserNotificationPolicyLogRecord
     from apps.user_management.models import User
 
@@ -198,6 +198,22 @@ def _create_notification_finished_user_notification_policy_log_record():
             log_record = _create_notification_finished_user_notification_policy_log_record()
             task_logger.info(f"Personal escalation exceeded. User: {user.pk}, alert_group: {alert_group.pk}")
         else:
+            if (
+                # don't force notify direct paged user who has already acknowledged the alert group
+                # after the last time they were paged.
+                notify_even_acknowledged
+                and (
+                    direct_paging_log_record := alert_group.log_records.filter(
+                        type=AlertGroupLogRecord.TYPE_DIRECT_PAGING, step_specific_info__user=user.public_primary_key
+                    ).last()
+                )
+                and alert_group.log_records.filter(
+                    type=AlertGroupLogRecord.TYPE_ACK, author=user, created_at__gte=direct_paging_log_record.created_at
+                ).exists()
+            ):
+                notify_even_acknowledged = False
+                task_logger.info(f"notify_even_acknowledged=False for user {user.pk}, alert_group {alert_group.pk})")
+
             if (
                 (alert_group.acknowledged and not notify_even_acknowledged)
                 or (alert_group.silenced and not notify_anyway)
diff --git a/engine/apps/alerts/tests/test_notify_user.py b/engine/apps/alerts/tests/test_notify_user.py
index 8de1e6792c..5c0d638178 100644
--- a/engine/apps/alerts/tests/test_notify_user.py
+++ b/engine/apps/alerts/tests/test_notify_user.py
@@ -5,6 +5,7 @@
 from telegram.error import RetryAfter
 
 from apps.alerts.models import AlertGroup
+from apps.alerts.paging import direct_paging
 from apps.alerts.tasks.notify_user import notify_user_task, perform_notification, send_bundled_notification
 from apps.api.permissions import LegacyAccessControlRole
 from apps.base.models.user_notification_policy import UserNotificationPolicy
@@ -577,3 +578,48 @@ def test_notify_user_task_notification_bundle_eta_is_outdated(
     assert notification_bundle.notification_task_id != test_task_id
     assert notification_bundle.last_notified_at == now
     assert notification_bundle.notifications.count() == 2
+
+
+@patch.object(perform_notification, "apply_async")
+@pytest.mark.django_db
+def test_notify_user_task_direct_paging_acknowledged(
+    mock_perform_notification_apply_async,
+    make_organization,
+    make_user,
+    make_alert_receive_channel,
+    make_alert_group,
+    make_user_notification_policy_log_record,
+    make_user_notification_policy,
+    django_capture_on_commit_callbacks,
+):
+    organization = make_organization()
+    from_user = make_user(organization=organization)
+    user1 = make_user(organization=organization)
+    user2 = make_user(organization=organization)
+    make_user_notification_policy(
+        user=user1,
+        step=UserNotificationPolicy.Step.NOTIFY,
+        notify_by=UserNotificationPolicy.NotificationChannel.TESTONLY,
+    )
+    make_user_notification_policy(
+        user=user2,
+        step=UserNotificationPolicy.Step.NOTIFY,
+        notify_by=UserNotificationPolicy.NotificationChannel.TESTONLY,
+    )
+    alert_receive_channel = make_alert_receive_channel(organization=organization)
+    alert_group = make_alert_group(alert_receive_channel=alert_receive_channel)
+
+    direct_paging(organization, from_user, "Test", alert_group=alert_group, users=[(user1, False), (user2, False)])
+    alert_group.acknowledge_by_user_or_backsync(user1)
+
+    # no notification should be sent for user1 because they have acknowledged the alert group
+    with django_capture_on_commit_callbacks(execute=True):
+        notify_user_task(user1.pk, alert_group.pk, notify_even_acknowledged=True)
+
+    mock_perform_notification_apply_async.assert_not_called()
+
+    # user2 should receive a notification because they have not acknowledged the alert group
+    with django_capture_on_commit_callbacks(execute=True):
+        notify_user_task(user2.pk, alert_group.pk, notify_even_acknowledged=True)
+
+    mock_perform_notification_apply_async.assert_called_once()